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

Issue:
Revisit BolDB Existing Schema
Existing schema doesn’t persist execution results. It needs to be revisited in general


Discussion / hints:

Fix patch:
diff --git a/pkg/compute/executor.go b/pkg/compute/executor.go
index 3afa7b5f04..f6d598822f 100644
--- a/pkg/compute/executor.go
+++ b/pkg/compute/executor.go
@@ -359,7 +359,7 @@ func (e *BaseExecutor) Run(ctx context.Context, state store.LocalExecutionState)
 	jobsCompleted.Add(ctx, 1)
 
 	expectedState := store.ExecutionStateRunning
-	publishedResult := models.SpecConfig{}
+	var publishedResult *models.SpecConfig
 
 	// publish if the job has a publisher defined
 	if !execution.Job.Task().Publisher.IsEmpty() {
@@ -368,6 +368,7 @@ func (e *BaseExecutor) Run(ctx context.Context, state store.LocalExecutionState)
 			ExecutionID:    execution.ID,
 			ExpectedStates: []store.LocalExecutionStateType{expectedState},
 			NewState:       store.ExecutionStatePublishing,
+			RunOutput:      result,
 		}); err != nil {
 			return err
 		}
@@ -395,10 +396,12 @@ func (e *BaseExecutor) Run(ctx context.Context, state store.LocalExecutionState)
 	}
 
 	// mark the execution as completed
-	if err := e.store.UpdateExecutionState(ctx, store.UpdateExecutionStateRequest{
-		ExecutionID:    execution.ID,
-		ExpectedStates: []store.LocalExecutionStateType{expectedState},
-		NewState:       store.ExecutionStateCompleted,
+	if err = e.store.UpdateExecutionState(ctx, store.UpdateExecutionStateRequest{
+		ExecutionID:     execution.ID,
+		ExpectedStates:  []store.LocalExecutionStateType{expectedState},
+		NewState:        store.ExecutionStateCompleted,
+		PublishedResult: publishedResult,
+		RunOutput:       result,
 	}); err != nil {
 		return err
 	}
@@ -410,7 +413,7 @@ func (e *BaseExecutor) Run(ctx context.Context, state store.LocalExecutionState)
 			SourcePeerID: e.ID,
 			TargetPeerID: state.RequesterNodeID,
 		},
-		PublishResult:    &publishedResult,
+		PublishResult:    publishedResult,
 		RunCommandResult: result,
 	})
 	return err
@@ -419,26 +422,24 @@ func (e *BaseExecutor) Run(ctx context.Context, state store.LocalExecutionState)
 // Publish the result of an execution after it has been verified.
 func (e *BaseExecutor) publish(ctx context.Context, localExecutionState store.LocalExecutionState,
 	resultFolder string,
-) (publishedResult models.SpecConfig, err error) {
+) (*models.SpecConfig, error) {
 	execution := localExecutionState.Execution
 	log.Ctx(ctx).Debug().Msgf("Publishing execution %s", execution.ID)
 
 	jobPublisher, err := e.publishers.Get(ctx, execution.Job.Task().Publisher.Type)
 	if err != nil {
-		err = fmt.Errorf("failed to get publisher %s: %w", execution.Job.Task().Publisher.Type, err)
-		return
+		return nil, fmt.Errorf("failed to get publisher %s: %w", execution.Job.Task().Publisher.Type, err)
 	}
-	publishedResult, err = jobPublisher.PublishResult(ctx, execution, resultFolder)
+	publishedResult, err := jobPublisher.PublishResult(ctx, execution, resultFolder)
 	if err != nil {
-		err = fmt.Errorf("failed to publish result: %w", err)
-		return
+		return nil, fmt.Errorf("failed to publish result: %w", err)
 	}
 
 	log.Ctx(ctx).Debug().
 		Str("execution", execution.ID).
 		Msg("Execution published")
 
-	return
+	return &publishedResult, nil
 }
 
 // Cancel the execution.
diff --git a/pkg/compute/store/boltdb/constants.go b/pkg/compute/store/boltdb/constants.go
new file mode 100644
index 0000000000..3c6210d08d
--- /dev/null
+++ b/pkg/compute/store/boltdb/constants.go
@@ -0,0 +1,41 @@
+package boltdb
+
+const (
+	defaultSliceRetrievalCapacity = 10
+	defaultDatabasePermissions    = 0600
+)
+
+// Schema versioning
+// The currentSchemaVersion is used as a prefix for all bucket names to support
+// future schema changes while maintaining backward compatibility. When the data
+// structure needs to be modified in a way that's not backward compatible, increment
+// this version and create new buckets with the updated schema.
+//
+// This approach allows for:
+// 1. Clear separation between different schema versions
+// 2. Easier data migrations (old version buckets can coexist with new ones)
+// 3. Gradual rollout of schema changes
+// 4. Ability to roll back to previous schema version if issues are encountered
+//
+// To introduce a new schema version:
+//  1. Increment the currentSchemaVersion (e.g., to "v2"). This will auto create new buckets.
+//  2. Implement a migration strategy to move data from old buckets to new ones
+//  3. The store should now read/write data from the new buckets
+const currentSchemaVersion = "v1"
+
+// Bucket names
+// All bucket names are prefixed with the current schema version to support
+// versioning and backward compatibility
+const (
+	// Main buckets
+	executionsBucket       = currentSchemaVersion + "_executions"
+	executionHistoryBucket = currentSchemaVersion + "_execution_history"
+
+	// Index buckets
+	idxExecutionsByJobBucket   = currentSchemaVersion + "_idx_executions_by_job_id"
+	idxExecutionsByStateBucket = currentSchemaVersion + "_idx_executions_by_state"
+
+	// Event-related buckets
+	eventsBucket      = currentSchemaVersion + "_events"
+	checkpointsBucket = currentSchemaVersion + "_checkpoints"
+)
diff --git a/pkg/compute/store/boltdb/counter.go b/pkg/compute/store/boltdb/counter.go
deleted file mode 100644
index 3d2f27c4e1..0000000000
--- a/pkg/compute/store/boltdb/counter.go
+++ /dev/null
@@ -1,50 +0,0 @@
-package boltdb
-
-import (
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
-	"go.uber.org/atomic"
-)
-
-type StateCounter struct {
-	data map[store.LocalExecutionStateType]*atomic.Uint64
-}
-
-func NewStateCounter() *StateCounter {
-	return &StateCounter{
-		data: make(map[store.LocalExecutionStateType]*atomic.Uint64),
-	}
-}
-
-func (s *StateCounter) IncrementState(key store.LocalExecutionStateType, amount uint64) {
-	counter, ok := s.data[key]
-	if !ok {
-		s.data[key] = atomic.NewUint64(amount)
-	} else {
-		counter.Add(amount)
-	}
-}
-
-func (s *StateCounter) DecrementState(key store.LocalExecutionStateType, amount uint64) {
-	counter, ok := s.data[key]
-	if !ok {
-		// We shouldn't get to the point where we have a missing state that
-		// we want to decrement, but to be defensive we will add it
-		s.data[key] = atomic.NewUint64(0)
-	} else {
-		counter.Sub(amount)
-	}
-}
-
-func (s *StateCounter) Include(other *StateCounter) {
-	for k, v := range other.data {
-		s.IncrementState(k, v.Load())
-	}
-}
-
-func (s *StateCounter) Get(key store.LocalExecutionStateType) uint64 {
-	v, ok := s.data[key]
-	if !ok {
-		return 0
-	}
-	return v.Load()
-}
diff --git a/pkg/compute/store/boltdb/database.go b/pkg/compute/store/boltdb/database.go
deleted file mode 100644
index 830b9e7248..0000000000
--- a/pkg/compute/store/boltdb/database.go
+++ /dev/null
@@ -1,19 +0,0 @@
-package boltdb
-
-import (
-	"time"
-
-	bolt "go.etcd.io/bbolt"
-)
-
-const (
-	DefaultDatabasePermissions = 0600
-)
-
-func GetDatabase(path string) (*bolt.DB, error) {
-	database, err := bolt.Open(path, DefaultDatabasePermissions, &bolt.Options{Timeout: 2 * time.Second})
-	if err != nil {
-		return nil, err
-	}
-	return database, nil
-}
diff --git a/pkg/compute/store/boltdb/store.go b/pkg/compute/store/boltdb/store.go
index 13226a87b3..03ce68f194 100644
--- a/pkg/compute/store/boltdb/store.go
+++ b/pkg/compute/store/boltdb/store.go
@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"reflect"
 	"sort"
-	"sync"
 	"time"
 
 	"github.com/rs/zerolog/log"
@@ -18,66 +17,59 @@ import (
 	boltdb_watcher "github.com/bacalhau-project/bacalhau/pkg/lib/watcher/boltdb"
 )
 
-const (
-	newExecutionComment = "LocalExecutionState created"
-
-	DefaultSliceRetrievalCapacity = 10
-
-	BucketExecutionsName = "execution"
-	BucketHistoryName    = "execution-history"
-	BucketJobIndexName   = "execution-index"
-	BucketLiveIndexName  = "execution-live-index"
-
-	BucketEventsName      = "events"
-	BucketCheckpointsName = "checkpoints"
-)
-
 // Store represents an execution store that is backed by a boltdb database
-// on disk.  The structure of the store is in boltdb buckets and sub-buckets
-// so that they can be created and retrieved easily.
+// on disk. The structure of the store is organized into boltdb buckets and sub-buckets
+// for efficient data retrieval and organization.
+//
+// The schema (<key> -> {json-value}) looks like the following, where <> represents
+// keys, {} represents values, and undecorated values are boltdb buckets:
 //
-// The schema (<key> {json-value}) looks like the following where <> represents
-// keys, {} represents values and undecorated values are boltdb buckets.
+// * Executions are stored in a bucket called `execution` where each key is
+// an execution ID and the value is the JSON representation of the execution.
 //
-// * Executions are stored in a bucket called `executions` where each key is
-// an execution ID and the value is the JSON representation.
+// executions
 //
-// execution
-//     |--> <execution-id> -> {store.LocalExecutionState}
+//	|--> <execution-id> -> {store.LocalExecutionState}
 //
-// * LocalExecutionState history is stored in a bucket called `history`. Each execution
+// * Execution history is stored in a bucket called `execution-history`. Each execution
 // that has history is stored in a sub-bucket, whose name is the execution ID.
-// Within the execution id bucket, each key is a sequential value for the
-// history item being written so they are returned in write-order
+// Within the execution ID bucket, each key is a sequential number for the
+// history item, ensuring they are returned in write order.
 //
-// execution-history
-//     |--> execution-id
-//               |-> <seqnum> -> {store.LocalStateHistory}
+// execution_history
 //
-// * The job index is stored in a bucket called execution-index where
-// each job is represented by a new bucket, named after the job ID.  Within
-// that job, each execution is represented by a key which is the id of that
-// execution, the value being nil.
+//	|--> <execution-id>
+//	          |--> <seqnum> -> {store.LocalStateHistory}
 //
-// execution-index
-//     |--> job-id
-//               |-> <execution-id> -> {}
+// * The job index is stored in a bucket called `idx:executions-by-jobid` where
+// each job is represented by a sub-bucket, named after the job ID. Within
+// that job bucket, each execution is represented by a key which is the ID of that
+// execution, with a nil value.
 //
-// * The live index is stored in a bucket called execution-live-index where each
-// execution that is in an active state (currently ExecutionStateBidAccepted).
-// This is used at node start to check what nodes _should_ be running.
+// idx_executions_by_job_id
 //
-// execution-live-index
-//    |-> <execution-id> -> {}
+//	|--> <job-id>
+//	          |--> <execution-id> -> nil
 //
-
+// * The state index is stored in a bucket called `idx:executions-by-state` where
+// each execution state is represented by a sub-bucket, named after the state.
+// Within each state bucket, executions are indexed by their ID with a nil value.
+//
+// idx_executions_by_state
+//
+//	|--> <state>
+//	          |--> <execution-id> -> nil
+//
+// * Additional buckets for event storage:
+//   - `events`: Stores event data
+//   - `checkpoints`: Stores checkpoint information for the event system
+//
+// This structure allows for efficient querying of executions by ID, job, and state,
+// as well as maintaining a complete history of execution state changes.
 type Store struct {
 	database   *bolt.DB
 	marshaller marshaller.Marshaller
-
-	starting     sync.WaitGroup
-	stateCounter *StateCounter
-	eventStore   *boltdb_watcher.EventStore
+	eventStore *boltdb_watcher.EventStore
 }
 
 // NewStore creates a new store backed by a boltdb database at the
@@ -87,48 +79,29 @@ type Store struct {
 // it would mean later transactions will fail unless they obtain their
 // own reference to the bucket.
 func NewStore(ctx context.Context, dbPath string) (*Store, error) {
-	s := &Store{
-		marshaller:   marshaller.NewJSONMarshaller(),
-		starting:     sync.WaitGroup{},
-		stateCounter: NewStateCounter(),
-	}
 	log.Ctx(ctx).Debug().Msgf("creating new bbolt database at %s", dbPath)
 
-	database, err := GetDatabase(dbPath)
+	database, err := bolt.Open(dbPath, defaultDatabasePermissions, &bolt.Options{Timeout: 2 * time.Second})
 	if err != nil {
 		if errors.Is(err, bolt.ErrTimeout) {
-			return nil, fmt.Errorf("timed out while opening database, is file %q in use?", dbPath)
+			return nil, fmt.Errorf("timed out while opening database, file %q might be in use", dbPath)
 		}
 		return nil, err
 	}
 
-	s.database = database
-	err = s.database.Update(func(tx *bolt.Tx) error {
-		_, err = tx.CreateBucketIfNotExists([]byte(BucketExecutionsName))
-		if err != nil {
-			return err
-		}
-
-		_, err = tx.CreateBucketIfNotExists([]byte(BucketHistoryName))
-		if err != nil {
-			return err
-		}
-
-		_, err = tx.CreateBucketIfNotExists([]byte(BucketJobIndexName))
-		if err != nil {
-			return err
-		}
-
-		_, err = tx.CreateBucketIfNotExists([]byte(BucketLiveIndexName))
-		if err != nil {
-			return err
+	err = database.Update(func(tx *bolt.Tx) error {
+		buckets := []string{executionsBucket, executionHistoryBucket, idxExecutionsByJobBucket, idxExecutionsByStateBucket}
+		for _, b := range buckets {
+			_, err = tx.CreateBucketIfNotExists(strToBytes(b))
+			if err != nil {
+				return fmt.Errorf("error creating bucket %s: %w", b, err)
+			}
 		}
-
 		return nil
 	})
 
 	if err != nil {
-		return nil, fmt.Errorf("error creating database structure: %s", err)
+		return nil, fmt.Errorf("error creating database structure: %w", err)
 	}
 
 	eventObjectSerializer := watcher.NewJSONSerializer()
@@ -136,84 +109,74 @@ func NewStore(ctx context.Context, dbPath string) (*Store, error) {
 	if err != nil {
 		return nil, fmt.Errorf("failed to register LocalStateHistory type: %w", err)
 	}
-	s.eventStore, err = boltdb_watcher.NewEventStore(database,
-		boltdb_watcher.WithEventsBucket(BucketEventsName),
-		boltdb_watcher.WithCheckpointBucket(BucketCheckpointsName),
+	eventStore, err := boltdb_watcher.NewEventStore(database,
+		boltdb_watcher.WithEventsBucket(eventsBucket),
+		boltdb_watcher.WithCheckpointBucket(checkpointsBucket),
 		boltdb_watcher.WithEventSerializer(eventObjectSerializer),
 	)
 
-	// Populate the state counter for the
-	s.starting.Add(1)
-	go s.populateStateCounter(ctx)
-
-	return s, nil
-}
-
-// getExecutionsBucket helper gets a reference to the executions bucket within
-// the supplied transaction.
-func (s *Store) getExecutionsBucket(tx *bolt.Tx) *bolt.Bucket {
-	return tx.Bucket([]byte(BucketExecutionsName))
-}
-
-// getHistoryBucket helper gets a reference to the execution history bucket
-// within the supplied transaction.
-func (s *Store) getHistoryBucket(tx *bolt.Tx) *bolt.Bucket {
-	return tx.Bucket([]byte(BucketHistoryName))
-}
-
-// getJobIndexBucket helper gets a reference to the job index bucket within
-// the supplied transaction.
-func (s *Store) getJobIndexBucket(tx *bolt.Tx) *bolt.Bucket {
-	return tx.Bucket([]byte(BucketJobIndexName))
-}
-
-// getLiveIndexBucket helper gets a reference to the live execution index bucket
-func (s *Store) getLiveIndexBucket(tx *bolt.Tx) *bolt.Bucket {
-	return tx.Bucket([]byte(BucketLiveIndexName))
+	return &Store{
+		database:   database,
+		marshaller: marshaller.NewJSONMarshaller(),
+		eventStore: eventStore,
+	}, nil
 }
 
 // GetExecution returns the stored LocalExecutionState structure for the provided execution ID.
 func (s *Store) GetExecution(ctx context.Context, executionID string) (store.LocalExecutionState, error) {
-	log.Ctx(ctx).Trace().
-		Str("ExecutionID", executionID).
-		Msg("boltdb.GetExecution")
-
 	var execution store.LocalExecutionState
-	err := s.database.View(func(tx *bolt.Tx) (err error) {
-		execution, err = s.getExecution(tx, executionID)
-		return
+	err := s.database.View(func(tx *bolt.Tx) error {
+		var err error
+		execution, err = s.getExecutionInTx(tx, executionID)
+		return err
 	})
-
-	execution.Normalize()
 	return execution, err
 }
 
-func (s *Store) getExecution(tx *bolt.Tx, executionID string) (store.LocalExecutionState, error) {
+func (s *Store) getExecutionInTx(tx *bolt.Tx, executionID string) (store.LocalExecutionState, error) {
 	var execution store.LocalExecutionState
 
-	executionsBucket := tx.Bucket([]byte(BucketExecutionsName))
-	data := executionsBucket.Get([]byte(executionID))
+	data := bucket(tx, executionsBucket).Get(strToBytes(executionID))
 	if data == nil {
 		return execution, store.NewErrExecutionNotFound(executionID)
 	}
 
 	err := s.marshaller.Unmarshal(data, &execution)
-	return execution, err
+	if err != nil {
+		return execution, fmt.Errorf("failed to unmarshal execution: %w", err)
+	}
+	execution.Normalize()
+	return execution, nil
 }
 
 // GetExecutions retrieves akk if the executions from the job-index bucket for the
 // provided Job ID.
 func (s *Store) GetExecutions(ctx context.Context, jobID string) ([]store.LocalExecutionState, error) {
-	log.Ctx(ctx).Trace().
-		Str("JobID", jobID).
-		Msg("boltdb.GetExecutions")
-
-	var executions []store.LocalExecutionState
-	err := s.database.View(func(tx *bolt.Tx) (err error) {
-		executions, err = s.getExecutions(tx, jobID)
-		return
+	executions := make([]store.LocalExecutionState, 0, defaultSliceRetrievalCapacity)
+
+	err := s.database.View(func(tx *bolt.Tx) error {
+		jobBucket := bucket(tx, idxExecutionsByJobBucket, jobID)
+		if jobBucket == nil {
+			return store.NewErrExecutionsNotFoundForJob(jobID)
+		}
+
+		// List all of the keys in the bucket which all have nil values, and are
+		// used only as markers to point to the actual execution in the relevant
+		// bucket
+		return jobBucket.ForEach(func(k, _ []byte) error {
+			execution, err := s.getExecutionInTx(tx, string(k))
+			if err != nil {
+				return err
+			}
+			executions = append(executions, execution)
+			return nil
+		})
 	})
-	if err != nil || len(executions) == 0 {
+
+	if err != nil {
+		return nil, err
+	}
+	if len(executions) == 0 {
 		return nil, store.NewErrExecutionsNotFoundForJob(jobID)
 	}
 
@@ -221,46 +184,36 @@ func (s *Store) GetExecutions(ctx context.Context, jobID string) ([]store.LocalE
 		return executions[i].UpdateTime.Before(executions[j].UpdateTime)
 	})
 
-	for _, execution := range executions {
-		execution.Normalize()
-	}
 	return executions, nil
 }
 
-func (s *Store) getExecutions(tx *bolt.Tx, jobID string) ([]store.LocalExecutionState, error) {
-	jobIndexBucket := s.getJobIndexBucket(tx)
-	jobBucket := jobIndexBucket.Bucket([]byte(jobID))
-	if jobBucket == nil {
-		return nil, store.NewErrJobNotFound(jobID)
-	}
-
-	executions := make([]store.LocalExecutionState, 0, DefaultSliceRetrievalCapacity)
+func (s *Store) GetLiveExecutions(ctx context.Context) ([]store.LocalExecutionState, error) {
+	executions := make([]store.LocalExecutionState, 0, defaultSliceRetrievalCapacity)
 
-	// List all of the keys in the bucket which all have nil values, and are
-	// used only as markers to point to the actual execution in the relevant
-	// bucket
-	err := jobBucket.ForEach(func(key []byte, _ []byte) error {
-		execution, err := s.getExecution(tx, string(key))
-		if err != nil {
-			return err
+	err := s.database.View(func(tx *bolt.Tx) error {
+		for _, state := range store.ExecutionStateTypes() {
+			if !state.IsExecuting() {
+				continue
+			}
+			stateBucket := bucket(tx, idxExecutionsByStateBucket, state.String())
+			if stateBucket == nil {
+				continue
+			}
+			err := stateBucket.ForEach(func(k, _ []byte) error {
+				execution, err := s.getExecutionInTx(tx, string(k))
+				if err != nil {
+					return err
+				}
+				executions = append(executions, execution)
+
+				return nil
+			})
+			if err != nil {
+				return err
+			}
 		}
-		executions = append(executions, execution)
-
 		return nil
 	})
-
-	return executions, err
-}
-
-func (s *Store) GetLiveExecutions(ctx context.Context) ([]store.LocalExecutionState, error) {
-	log.Ctx(ctx).Trace().
-		Msg("boltdb.GetLiveExecutions")
-
-	var executions []store.LocalExecutionState
-	err := s.database.View(func(tx *bolt.Tx) (err error) {
-		executions, err = s.getLiveExecutions(tx)
-		return
-	})
 	if err != nil {
 		return nil, err
 	}
@@ -272,298 +225,218 @@ func (s *Store) GetLiveExecutions(ctx context.Context) ([]store.LocalExecutionSt
 	return executions, nil
 }
 
-func (s *Store) getLiveExecutions(tx *bolt.Tx) ([]store.LocalExecutionState, error) {
-	liveIndexBkt := s.getLiveIndexBucket(tx)
-
-	executions := make([]store.LocalExecutionState, 0, DefaultSliceRetrievalCapacity)
+// GetExecutionHistory retrieves the execution history for a single execution
+// specified by the executionID parameter.
+func (s *Store) GetExecutionHistory(ctx context.Context, executionID string) ([]store.LocalStateHistory, error) {
+	history := make([]store.LocalStateHistory, 0, defaultSliceRetrievalCapacity)
 
-	// List all of the keys in the live index bucket, and fetch the appropriate
-	// executions
-	err := liveIndexBkt.ForEach(func(key []byte, _ []byte) error {
-		execution, err := s.getExecution(tx, string(key))
-		if err != nil {
-			return err
+	err := s.database.View(func(tx *bolt.Tx) error {
+		historyBucket := bucket(tx, executionHistoryBucket, executionID)
+		if historyBucket == nil {
+			return store.NewErrExecutionHistoryNotFound(executionID)
 		}
-		executions = append(executions, execution)
 
-		return nil
-	})
-	return executions, err
-}
+		// Iterate all of the key-values in the history/executionID bucket as they
+		// are the history, and are written in sequential order using the bucket
+		// sequence
+		return historyBucket.ForEach(func(k, v []byte) error {
+			var historyItem store.LocalStateHistory
+			err := s.marshaller.Unmarshal(v, &historyItem)
+			if err != nil {
+				return err
+			}
 
-// GetExecutionHistory retrieves the execution history for a single execution
-// specified by the executionID parameter.
-func (s *Store) GetExecutionHistory(ctx context.Context, executionID string) ([]store.LocalStateHistory, error) {
-	log.Ctx(ctx).Trace().
-		Str("ExecutionID", executionID).
-		Msg("boltdb.GetExecutionHistory")
-
-	var history []store.LocalStateHistory
-	err := s.database.View(func(tx *bolt.Tx) (err error) {
-		history, err = s.getExecutionHistory(tx, executionID)
-		return
+			history = append(history, historyItem)
+			return nil
+		})
 	})
-	if err != nil || len(history) == 0 {
+
+	if err != nil {
+		return nil, err
+	}
+	if len(history) == 0 {
 		return nil, store.NewErrExecutionHistoryNotFound(executionID)
 	}
 
-	return history, err
+	return history, nil
 }
 
-func (s *Store) getExecutionHistory(tx *bolt.Tx, executionID string) ([]store.LocalStateHistory, error) {
-	historyBucket := s.getHistoryBucket(tx)
-
-	executionBucket := historyBucket.Bucket([]byte(executionID))
-	if executionBucket == nil {
-		return nil, store.NewErrExecutionNotFound(executionID)
+func (s *Store) CreateExecution(ctx context.Context, execution store.LocalExecutionState) error {
+	execution.Normalize()
+	if err := store.ValidateNewExecution(execution); err != nil {
+		return fmt.Errorf("CreateExecution failure: %w", err)
 	}
 
-	history := make([]store.LocalStateHistory, 0, DefaultSliceRetrievalCapacity)
-
-	// Iterate all of the key-values in the history/executionID bucket as they
-	// are the history, and are written in sequential order using the bucket
-	// sequence
-	err := executionBucket.ForEach(func(key []byte, data []byte) error {
-		var item store.LocalStateHistory
+	return s.database.Update(func(tx *bolt.Tx) error {
+		// Check if execution already exists
+		_, err := s.getExecutionInTx(tx, execution.Execution.ID)
+		if err == nil {
+			return store.NewErrExecutionAlreadyExists(execution.Execution.ID)
+		} else if !errors.As(err, &store.ErrExecutionNotFound{}) {
+			return err
+		}
 
-		err := s.marshaller.Unmarshal(data, &item)
+		// Store execution
+		executionData, err := s.marshaller.Marshal(execution)
 		if err != nil {
 			return err
 		}
+		if err := bucket(tx, executionsBucket).Put(strToBytes(execution.Execution.ID), executionData); err != nil {
+			return err
+		}
 
-		history = append(history, item)
-		return nil
-	})
-
-	return history, err
-}
-
-// CreateExecution creates a new execution in the database and also sets up the
-// relevant index entry for the new execution.
-func (s *Store) CreateExecution(ctx context.Context, localExecutionState store.LocalExecutionState) error {
-	log.Ctx(ctx).Trace().
-		Str("ExecutionID", localExecutionState.Execution.ID).
-		Msg("boltdb.CreateExecution")
-
-	localExecutionState.Normalize()
-	if err := store.ValidateNewExecution(localExecutionState); err != nil {
-		return fmt.Errorf("CreateExecution failure: %w", err)
-	}
+		// Create job-execution index
+		jobBucket, err := bucket(tx, idxExecutionsByJobBucket).CreateBucketIfNotExists(strToBytes(execution.Execution.JobID))
+		if err != nil {
+			return err
+		}
+		if err := jobBucket.Put(strToBytes(execution.Execution.ID), nil); err != nil {
+			return err
+		}
 
-	return s.database.Update(func(tx *bolt.Tx) (err error) {
-		err = s.createExecution(tx, localExecutionState)
-		if err == nil {
-			// If we are confident that the value was written without error
-			// and we won't rollback
-			s.stateCounter.IncrementState(localExecutionState.State, 1)
+		// Create execution-state index
+		stateBucket, err := bucket(tx, idxExecutionsByStateBucket).CreateBucketIfNotExists(strToBytes(execution.State.String()))
+		if err != nil {
+			return err
+		}
+		if err := stateBucket.Put(strToBytes(execution.Execution.ID), nil); err != nil {
+			return err
 		}
 
-		return
+		// Append to history
+		return s.appendHistory(tx, execution, store.ExecutionStateUndefined, store.NewExecutionMessage)
 	})
 }
 
-func (s *Store) createExecution(tx *bolt.Tx, localExecutionState store.LocalExecutionState) error {
-	_, err := s.getExecution(tx, localExecutionState.Execution.ID)
-	if err == nil { // deliberate, we require an err to continue
-		return store.NewErrExecutionAlreadyExists(localExecutionState.Execution.ID)
-	}
-
-	// Write the execution to the executions bucket
-	executionData, err := s.marshaller.Marshal(localExecutionState)
-	if err != nil {
-		return err
-	}
-
-	executionsBucket := s.getExecutionsBucket(tx)
-	err = executionsBucket.Put([]byte(localExecutionState.Execution.ID), executionData)
-	if err != nil {
-		return err
-	}
-
-	// Create the job bucket in the job index if it does not already exist
-	jobIndexBucket := s.getJobIndexBucket(tx)
-	jobBucket, err := jobIndexBucket.CreateBucketIfNotExists([]byte(localExecutionState.Execution.JobID))
-	if err != nil {
-		return err
-	}
-
-	err = jobBucket.Put([]byte(localExecutionState.Execution.ID), nil)
-	if err != nil {
-		return err
-	}
-
-	return s.appendHistory(tx, localExecutionState, store.ExecutionStateUndefined, newExecutionComment)
-}
-
-// UpdateExecutionState updates the current state of the execution
 func (s *Store) UpdateExecutionState(ctx context.Context, request store.UpdateExecutionStateRequest) error {
-	log.Ctx(ctx).Trace().
-		Str("ExecutionID", request.ExecutionID).
-		Msg("boltdb.UpdateExecutionState")
-
-	return s.database.Update(func(tx *bolt.Tx) (err error) {
-		previousState, err := s.updateExecutionState(tx, request)
-		if err == nil {
-			s.stateCounter.DecrementState(previousState, 1)
-			s.stateCounter.IncrementState(request.NewState, 1)
+	return s.database.Update(func(tx *bolt.Tx) error {
+		execution, err := s.getExecutionInTx(tx, request.ExecutionID)
+		if err != nil {
+			return err
 		}
-		return err
-	})
-}
 
-func (s *Store) updateExecutionState(tx *bolt.Tx, request store.UpdateExecutionStateRequest) (store.LocalExecutionStateType, error) {
-	emptyState := store.ExecutionStateUndefined
-
-	localExecutionState, err := s.getExecution(tx, request.ExecutionID)
-	if err != nil {
-		return emptyState, store.NewErrExecutionNotFound(request.ExecutionID)
-	}
-
-	if err := request.Validate(localExecutionState); err != nil {
-		return emptyState, err
-	}
-
-	if localExecutionState.State.IsTerminal() {
-		return emptyState, store.NewErrExecutionAlreadyTerminal(request.ExecutionID, localExecutionState.State, request.NewState)
-	}
-
-	previousState := localExecutionState.State
-	localExecutionState.State = request.NewState
-	localExecutionState.Version += 1
-	localExecutionState.UpdateTime = time.Now()
+		if err := request.Validate(execution); err != nil {
+			return err
+		}
 
-	// Having modified the execution, we're going to write it back over the top of the previous entry
-	// before appending a copy of the history to the history bucket
+		previousState := execution.State
+		execution.State = request.NewState
+		execution.Revision++
+		execution.UpdateTime = time.Now().UTC()
 
-	data, err := s.marshaller.Marshal(localExecutionState)
-	if err != nil {
-		return emptyState, err
-	}
-
-	executionsBucket := s.getExecutionsBucket(tx)
-	err = executionsBucket.Put([]byte(localExecutionState.Execution.ID), data)
-	if err != nil {
-		return emptyState, err
-	}
+		if request.RunOutput != nil {
+			execution.RunOutput = request.RunOutput
+		}
 
-	// If this execution is in an active state, then we should index it so that we know
-	// at a restart that it should be running. If it is not that, then we should ensure
-	// that the index for live executions does not include this ID.
-	var indexError error
-	indexKey := []byte(localExecutionState.Execution.ID)
-	bkt := s.getLiveIndexBucket(tx)
-
-	if localExecutionState.State.IsExecuting() {
-		// We can safely add an index key here even if it currently exists, so
-		// we don't need to check the previous states to see if it already
-		// exists.
-		indexError = bkt.Put(indexKey, []byte{})
-	} else {
-		// Removes the index key from the bucket, and if it doesn't exist then
-		// quietly returns nil as expected.
-		indexError = bkt.Delete(indexKey)
-	}
+		if request.PublishedResult != nil {
+			execution.PublishedResult = request.PublishedResult
+		}
 
-	if indexError != nil {
-		return previousState, fmt.Errorf("failed to process live index: %s", indexError)
-	}
+		// Update execution
+		executionData, err := s.marshaller.Marshal(execution)
+		if err != nil {
+			return err
+		}
+		if err = bucket(tx, executionsBucket).Put(strToBytes(execution.Execution.ID), executionData); err != nil {
+			return err
+		}
 
-	// Update the history of the execution
-	historyErr := s.appendHistory(tx, localExecutionState, previousState, request.Comment)
-	if historyErr != nil {
-		return previousState, fmt.Errorf("failed to append execution history: %s", historyErr)
-	}
+		// Update execution-state index
+		if err = bucket(tx, idxExecutionsByStateBucket, previousState.String()).Delete(strToBytes(execution.Execution.ID)); err != nil {
+			return err
+		}
+		stateBucket, err := bucket(tx, idxExecutionsByStateBucket).CreateBucketIfNotExists(strToBytes(execution.State.String()))
+		if err != nil {
+			return err
+		}
+		if err = stateBucket.Put(strToBytes(execution.Execution.ID), nil); err != nil {
+			return err
+		}
 
-	return previousState, nil
+		// Append to history
+		return s.appendHistory(tx, execution, previousState, request.Comment)
+	})
 }
 
-// Must be called where tx is a write transaction and adds a new history entry
-// to the bucket in history/execution-id/ with a key value derived from the
-// bucket sequence. Elsewhere we want to iterate through the history so to
-// make sure we get it back in creation order we use a three-digit sequence
-// number as bucket traversals happen in lexicographical order.
-func (s *Store) appendHistory(
-	tx *bolt.Tx,
-	updatedExecution store.LocalExecutionState,
-	previousState store.LocalExecutionStateType, comment string) error {
-	historyBucket := s.getHistoryBucket(tx)
-	executionHistoryBucket, err := historyBucket.CreateBucketIfNotExists([]byte(updatedExecution.Execution.ID))
-	if err != nil {
-		return err
-	}
-
+func (s *Store) appendHistory(tx *bolt.Tx,
+	execution store.LocalExecutionState, previousState store.LocalExecutionStateType, comment string) error {
 	historyEntry := store.LocalStateHistory{
-		ExecutionID:   updatedExecution.Execution.ID,
+		ExecutionID:   execution.Execution.ID,
 		PreviousState: previousState,
-		NewState:      updatedExecution.State,
-		NewVersion:    updatedExecution.Version,
+		NewState:      execution.State,
+		NewRevision:   execution.Revision,
 		Comment:       comment,
-		Time:          updatedExecution.UpdateTime,
+		Time:          execution.UpdateTime,
 	}
-	historyEntryData, err := s.marshaller.Marshal(historyEntry)
+	historyData, err := s.marshaller.Marshal(historyEntry)
 	if err != nil {
 		return err
 	}
 
-	// History entry keys within the execution bucket will be iterated over
-	// in lexographical order so we want to pad the numbers with leading 0s
-	// although this adds a hard limit of 999 history items for an execution
-	seqNum, err := executionHistoryBucket.NextSequence()
+	historyBucket, err := bucket(tx, executionHistoryBucket).CreateBucketIfNotExists(strToBytes(execution.Execution.ID))
+	if err != nil {
+		return err
+	}
+
+	seqNum, err := historyBucket.NextSequence()
 	if err != nil {
 		return err
 	}
-	seq := fmt.Sprintf("%03d", seqNum)
 
 	err = s.eventStore.StoreEventTx(tx, watcher.OperationCreate, "LocalStateHistory", historyEntry)
 	if err != nil {
 		return err
 	}
-	return executionHistoryBucket.Put([]byte(seq), historyEntryData)
+
+	return historyBucket.Put(uint64ToBytes(seqNum), historyData)
 }
 
 // DeleteExecution delete the execution, removes its history and removes it
 // from the job index (along with the job indexes bucket)
 func (s *Store) DeleteExecution(ctx context.Context, executionID string) error {
-	log.Ctx(ctx).Trace().
-		Str("ExecutionID", executionID).
-		Msg("boltdb.DeleteExecution")
+	return s.database.Update(func(tx *bolt.Tx) error {
+		execution, err := s.getExecutionInTx(tx, executionID)
+		if err != nil {
+			return err
+		}
 
-	return s.database.Update(func(tx *bolt.Tx) (err error) {
-		return s.deleteExecution(tx, executionID)
-	})
-}
+		// Delete execution
+		if err := bucket(tx, executionsBucket).Delete(strToBytes(executionID)); err != nil {
+			return fmt.Errorf("failed to delete execution: %w", err)
+		}
 
-func (s *Store) deleteExecution(tx *bolt.Tx, executionID string) error {
-	localExecutionState, err := s.getExecution(tx, executionID)
-	if err != nil {
-		return err
-	}
+		// Delete from job-execution index
+		jobBucket := bucket(tx, idxExecutionsByJobBucket, execution.Execution.JobID)
+		if jobBucket == nil {
+			log.Ctx(ctx).Warn().Msgf("job bucket not found while deleting execution %s", executionID)
+		} else {
+			if err = jobBucket.Delete(strToBytes(executionID)); err != nil {
+				return fmt.Errorf("failed to delete execution from job index: %w", err)
+			}
 
-	// Delete single execution entry
-	executionsBucket := s.getExecutionsBucket(tx)
-	err = executionsBucket.Delete([]byte(executionID))
-	if err != nil {
-		return err
-	}
+			// Check if job bucket is empty
+			if isEmptyBucket(jobBucket) {
+				if err = bucket(tx, idxExecutionsByJobBucket).DeleteBucket(strToBytes(execution.Execution.JobID)); err != nil {
+					return fmt.Errorf("failed to delete empty job bucket: %w", err)
+				}
+			}
+		}
 
-	// Delete from job index
-	jobIndexBucket := s.getJobIndexBucket(tx)
-	jobBucket := jobIndexBucket.Bucket([]byte(localExecutionState.Execution.JobID))
-	if jobBucket != nil {
-		err = jobBucket.Delete([]byte(executionID))
-		if err != nil {
-			return err
+		// Delete from execution-state index
+		stateBucket := bucket(tx, idxExecutionsByStateBucket, execution.State.String())
+		if stateBucket != nil {
+			if err := stateBucket.Delete(strToBytes(executionID)); err != nil {
+				return fmt.Errorf("failed to delete execution from state index: %w", err)
+			}
 		}
-	}
 
-	// Delete the bucket with the execution-id within the history bucket
-	historyBucket := s.getHistoryBucket(tx)
-	err = historyBucket.DeleteBucket([]byte(executionID))
-	if err != nil {
-		return err
-	}
+		// Delete history
+		if err = bucket(tx, executionHistoryBucket).DeleteBucket(strToBytes(executionID)); err != nil {
+			return fmt.Errorf("failed to delete execution history: %w", err)
+		}
 
-	return nil
+		return nil
+	})
 }
 
 // GetEventStore returns the event store for the execution store
@@ -571,55 +444,26 @@ func (s *Store) GetEventStore() watcher.EventStore {
 	return s.eventStore
 }
 
-// Close ensures the database is closed cleanly
 func (s *Store) Close(ctx context.Context) error {
-	return s.database.Close()
+	if s.database != nil {
+		return s.database.Close()
+	}
+	return nil
 }
 
 func (s *Store) GetExecutionCount(ctx context.Context, state store.LocalExecutionStateType) (uint64, error) {
-	log.Ctx(ctx).Trace().
-		Msg("boltdb.GetExecutionCount")
-
-	// We have to wait here to ensure the counter has been populated,
-	// so we will wait until we know it has finished
-	s.starting.Wait()
-	return s.stateCounter.Get(state), nil
-}
-
-func (s *Store) populateStateCounter(ctx context.Context) {
-	acc := NewStateCounter()
-
-	err := s.database.View(func(tx *bolt.Tx) (err error) {
-		bucket := s.getExecutionsBucket(tx)
-		cursor := bucket.Cursor()
-
-		for k, v := cursor.First(); k != nil; k, v = cursor.Next() {
-			if v == nil { // We can't do much with empty values
-				continue
-			}
-
-			var entry store.LocalExecutionState
-			err = s.marshaller.Unmarshal(v, &entry)
-			if err != nil {
-				return
-			}
-
-			acc.IncrementState(entry.State, 1)
+	var count uint64
+	err := s.database.View(func(tx *bolt.Tx) error {
+		b := bucket(tx, idxExecutionsByStateBucket, state.String())
+		if b == nil {
+			return nil
 		}
 
-		return err
+		count = uint64(b.Stats().KeyN)
+		return nil
 	})
 
-	if err != nil {
-		log.Ctx(ctx).Error().Err(err).Msg("failed to generate state counter for execution store")
-	}
-
-	// As more items may have been added whilst we were running, merge the
-	// statecounter we have generated into the existing one for the store
-	s.stateCounter.Include(acc)
-
-	log.Ctx(ctx).Trace().Msg("finished populating state counter")
-	s.starting.Done()
+	return count, err
 }
 
 // compile-time check that we implement the interface ExecutionStore
diff --git a/pkg/compute/store/boltdb/utils.go b/pkg/compute/store/boltdb/utils.go
new file mode 100644
index 0000000000..0bcee76b5b
--- /dev/null
+++ b/pkg/compute/store/boltdb/utils.go
@@ -0,0 +1,41 @@
+package boltdb
+
+import (
+	"encoding/binary"
+
+	bolt "go.etcd.io/bbolt"
+)
+
+// strToBytes converts a string to a byte slice
+func strToBytes(s string) []byte {
+	return []byte(s)
+}
+
+// uint64ToBytes converts an uint64 to a byte slice
+func uint64ToBytes(i uint64) []byte {
+	buf := make([]byte, 8) //nolint:gomnd
+	binary.BigEndian.PutUint64(buf, i)
+	return buf
+}
+
+// Helper function to safely access buckets and nested buckets
+func bucket(tx *bolt.Tx, bucketNames ...string) *bolt.Bucket {
+	var b *bolt.Bucket
+	for _, name := range bucketNames {
+		if b == nil {
+			b = tx.Bucket([]byte(name))
+		} else {
+			b = b.Bucket([]byte(name))
+		}
+		if b == nil {
+			return nil
+		}
+	}
+	return b
+}
+
+// isEmptyBucket checks if a bucket is empty
+func isEmptyBucket(b *bolt.Bucket) bool {
+	k, _ := b.Cursor().First()
+	return k == nil
+}
diff --git a/pkg/compute/store/errors.go b/pkg/compute/store/errors.go
index 4959a77d22..968aa089ec 100644
--- a/pkg/compute/store/errors.go
+++ b/pkg/compute/store/errors.go
@@ -39,19 +39,6 @@ func (e ErrExecutionsNotFoundForJob) Error() string {
 	return "no executions found for job: " + e.JobID
 }
 
-// ErrJobNotFound is returned when the a job isn't found when trying to get its executions
-type ErrJobNotFound struct {
-	JobID string
-}
-
-func NewErrJobNotFound(id string) ErrJobNotFound {
-	return ErrJobNotFound{JobID: id}
-}
-
-func (e ErrJobNotFound) Error() string {
-	return "job not found: " + e.JobID
-}
-
 // ErrExecutionHistoryNotFound is returned when the execution is not found
 type ErrExecutionHistoryNotFound struct {
 	ExecutionID string
@@ -93,19 +80,19 @@ func (e ErrInvalidExecutionState) Error() string {
 	return fmt.Sprintf("execution %s is in state %s but expected one of %v", e.ExecutionID, e.Actual, e.Expected)
 }
 
-// ErrInvalidExecutionVersion is returned when an execution has an invalid version.
-type ErrInvalidExecutionVersion struct {
+// ErrInvalidExecutionRevision is returned when an execution has an invalid revision.
+type ErrInvalidExecutionRevision struct {
 	ExecutionID string
 	Actual      int
 	Expected    int
 }
 
-func NewErrInvalidExecutionVersion(id string, actual int, expected int) ErrInvalidExecutionVersion {
-	return ErrInvalidExecutionVersion{ExecutionID: id, Actual: actual, Expected: expected}
+func NewErrInvalidExecutionRevision(id string, actual int, expected int) ErrInvalidExecutionRevision {
+	return ErrInvalidExecutionRevision{ExecutionID: id, Actual: actual, Expected: expected}
 }
 
-func (e ErrInvalidExecutionVersion) Error() string {
-	return fmt.Sprintf("execution %s has version %d but expected %d", e.ExecutionID, e.Actual, e.Expected)
+func (e ErrInvalidExecutionRevision) Error() string {
+	return fmt.Sprintf("execution %s has revision %d but expected %d", e.ExecutionID, e.Actual, e.Expected)
 }
 
 // ErrExecutionAlreadyTerminal is returned when an execution is already in terminal state and cannot be updated.
diff --git a/pkg/compute/store/events.go b/pkg/compute/store/events.go
new file mode 100644
index 0000000000..9a50030f97
--- /dev/null
+++ b/pkg/compute/store/events.go
@@ -0,0 +1,5 @@
+package store
+
+const (
+	NewExecutionMessage = "Execution created"
+)
diff --git a/pkg/compute/store/types.go b/pkg/compute/store/types.go
index e908b340dc..63484c4aef 100644
--- a/pkg/compute/store/types.go
+++ b/pkg/compute/store/types.go
@@ -12,22 +12,25 @@ import (
 
 type LocalExecutionState struct {
 	Execution       *models.Execution
+	PublishedResult *models.SpecConfig
+	RunOutput       *models.RunCommandResult
 	RequesterNodeID string
 	State           LocalExecutionStateType
-	Version         int
+	Revision        int
 	CreateTime      time.Time
 	UpdateTime      time.Time
 	LatestComment   string
 }
 
 func NewLocalExecutionState(execution *models.Execution, requesterNodeID string) *LocalExecutionState {
+	now := time.Now().UTC()
 	return &LocalExecutionState{
 		Execution:       execution,
 		RequesterNodeID: requesterNodeID,
 		State:           ExecutionStateCreated,
-		Version:         1,
-		CreateTime:      time.Now().UTC(),
-		UpdateTime:      time.Now().UTC(),
+		Revision:        1,
+		CreateTime:      now,
+		UpdateTime:      now,
 	}
 }
 
@@ -58,7 +61,7 @@ type LocalStateHistory struct {
 	ExecutionID   string
 	PreviousState LocalExecutionStateType
 	NewState      LocalExecutionStateType
-	NewVersion    int
+	NewRevision   int
 	Comment       string
 	Time          time.Time
 }
@@ -72,11 +75,13 @@ type ExecutionSummary struct {
 }
 
 type UpdateExecutionStateRequest struct {
-	ExecutionID     string
-	NewState        LocalExecutionStateType
-	ExpectedStates  []LocalExecutionStateType
-	ExpectedVersion int
-	Comment         string
+	ExecutionID      string
+	NewState         LocalExecutionStateType
+	ExpectedStates   []LocalExecutionStateType
+	ExpectedRevision int
+	PublishedResult  *models.SpecConfig
+	RunOutput        *models.RunCommandResult
+	Comment          string
 }
 
 // Validate checks if the condition matches the given execution
@@ -95,8 +100,8 @@ func (condition UpdateExecutionStateRequest) Validate(localExecutionState LocalE
 		}
 	}
 
-	if condition.ExpectedVersion != 0 && condition.ExpectedVersion != localExecutionState.Version {
-		return NewErrInvalidExecutionVersion(execution.ID, localExecutionState.Version, condition.ExpectedVersion)
+	if condition.ExpectedRevision != 0 && condition.ExpectedRevision != localExecutionState.Revision {
+		return NewErrInvalidExecutionRevision(execution.ID, localExecutionState.Revision, condition.ExpectedRevision)
 	}
 	return nil
 }
diff --git a/pkg/compute/store/utils.go b/pkg/compute/store/utils.go
index f4b4bcc712..094ce6de4b 100644
--- a/pkg/compute/store/utils.go
+++ b/pkg/compute/store/utils.go
@@ -1,46 +1,13 @@
 package store
 
-import (
-	"context"
-
-	"github.com/rs/zerolog/log"
-)
-
-// GetActiveExecution returns the active execution for a given job.
-// In case of a bug where we have more than a single active execution, the latest one is returned
-func GetActiveExecution(ctx context.Context, s ExecutionStore, jobID string) (LocalExecutionState, error) {
-	executions, err := s.GetExecutions(ctx, jobID)
-	if err != nil {
-		return LocalExecutionState{}, err
-	}
-
-	var activeExecution LocalExecutionState
-	var activeExecutionsCount int
-	for _, execution := range executions {
-		if execution.State.IsActive() {
-			activeExecutionsCount++
-			if activeExecutionsCount != 1 || execution.UpdateTime.After(activeExecution.UpdateTime) {
-				activeExecution = execution
-			}
-		}
-	}
-
-	if activeExecutionsCount > 1 {
-		log.Ctx(ctx).Warn().Msgf(
-			"Found %d active executions for job %s. Selecting the latest one", activeExecutionsCount, jobID)
-	}
-
-	return activeExecution, nil
-}
-
 func ValidateNewExecution(localExecutionState LocalExecutionState) error {
 	// state must be either created, or bid accepted if the execution is pre-approved
 	if localExecutionState.State != ExecutionStateCreated && localExecutionState.State != ExecutionStateBidAccepted {
 		return NewErrInvalidExecutionState(
 			localExecutionState.Execution.ID, localExecutionState.State, ExecutionStateCreated, ExecutionStateBidAccepted)
 	}
-	if localExecutionState.Version != 1 {
-		return NewErrInvalidExecutionVersion(localExecutionState.Execution.ID, localExecutionState.Version, 1)
+	if localExecutionState.Revision != 1 {
+		return NewErrInvalidExecutionRevision(localExecutionState.Execution.ID, localExecutionState.Revision, 1)
 	}
 
 	return nil
diff --git a/pkg/models/execution.go b/pkg/models/execution.go
index 9979e6cdb3..ab54ebfee5 100644
--- a/pkg/models/execution.go
+++ b/pkg/models/execution.go
@@ -156,6 +156,9 @@ func (e *Execution) Normalize() {
 			Tasks: make(map[string]*Resources),
 		}
 	}
+	if e.PublishedResult == nil {
+		e.PublishedResult = &SpecConfig{}
+	}
 	e.Job.Normalize()
 }
 
diff --git a/pkg/node/compute.go b/pkg/node/compute.go
index a7705d0d56..62e5ad4346 100644
--- a/pkg/node/compute.go
+++ b/pkg/node/compute.go
@@ -185,8 +185,6 @@ func NewComputeNode(
 	// register compute public http apis
 	compute_endpoint.NewEndpoint(compute_endpoint.EndpointParams{
 		Router:             apiServer.Router,
-		Bidder:             bidder,
-		Store:              executionStore,
 		DebugInfoProviders: debugInfoProviders,
 	})
 
diff --git a/pkg/publicapi/endpoint/compute/endpoint.go b/pkg/publicapi/endpoint/compute/endpoint.go
index 496588101a..6c40340ffe 100644
--- a/pkg/publicapi/endpoint/compute/endpoint.go
+++ b/pkg/publicapi/endpoint/compute/endpoint.go
@@ -3,37 +3,28 @@ package compute
 import (
 	"github.com/labstack/echo/v4"
 
-	"github.com/bacalhau-project/bacalhau/pkg/compute"
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/publicapi/middleware"
 )
 
 type EndpointParams struct {
 	Router             *echo.Echo
-	Bidder             compute.Bidder
-	Store              store.ExecutionStore
 	DebugInfoProviders []models.DebugInfoProvider
 }
 
 type Endpoint struct {
 	router             *echo.Echo
-	bidder             compute.Bidder
-	store              store.ExecutionStore
 	debugInfoProviders []models.DebugInfoProvider
 }
 
 func NewEndpoint(params EndpointParams) *Endpoint {
 	e := &Endpoint{
 		router:             params.Router,
-		bidder:             params.Bidder,
-		store:              params.Store,
 		debugInfoProviders: params.DebugInfoProviders,
 	}
 
 	g := e.router.Group("/api/v1/compute")
 	g.Use(middleware.SetContentType(echo.MIMEApplicationJSON))
 	g.POST("/debug", e.debug)
-	g.POST("/approve", e.approve)
 	return e
 }
diff --git a/pkg/publicapi/endpoint/compute/endpoint_approve.go b/pkg/publicapi/endpoint/compute/endpoint_approve.go
deleted file mode 100644
index 585671c246..0000000000
--- a/pkg/publicapi/endpoint/compute/endpoint_approve.go
+++ /dev/null
@@ -1,45 +0,0 @@
-package compute
-
-import (
-	"errors"
-	"net/http"
-	"os"
-
-	"github.com/bacalhau-project/bacalhau/pkg/bidstrategy"
-	"github.com/bacalhau-project/bacalhau/pkg/publicapi/signatures"
-	"github.com/labstack/echo/v4"
-)
-
-// approve godoc
-//
-//	@ID			apiServer/approver
-//	@Summary	Approves a job to be run on this compute node.
-//	@Tags		Compute Node
-//	@Produce	json
-//	@Success	200	{object}	string
-//	@Failure	400	{object}	string
-//	@Failure	403	{object}	string
-//	@Failure	500	{object}	string
-//	@Router		/api/v1/compute/approve [get]
-func (s *Endpoint) approve(c echo.Context) error {
-	request, err := signatures.UnmarshalSigned[bidstrategy.ModerateJobRequest](c.Request().Context(), c.Request().Body)
-	if err != nil {
-		return echo.NewHTTPError(http.StatusBadRequest, err.Error())
-	}
-
-	approvingClient := os.Getenv("BACALHAU_JOB_APPROVER")
-	if request.ClientID != approvingClient {
-		return echo.NewHTTPError(http.StatusUnauthorized, errors.New("approval submitted by unknown client"))
-	}
-
-	executions, err := s.store.GetExecutions(c.Request().Context(), request.JobID)
-	if err != nil {
-		return echo.NewHTTPError(http.StatusInternalServerError, err.Error())
-	}
-
-	for _, execution := range executions {
-		go s.bidder.ReturnBidResult(c.Request().Context(), execution, &request.Response)
-	}
-
-	return c.JSON(http.StatusOK, "Job approved.")
-}


Test patch:
diff --git a/pkg/compute/store/boltdb/store_test.go b/pkg/compute/store/boltdb/store_test.go
index 30b08fbf3a..afa4c1ab7e 100644
--- a/pkg/compute/store/boltdb/store_test.go
+++ b/pkg/compute/store/boltdb/store_test.go
@@ -4,441 +4,16 @@ package boltdb
 
 import (
 	"context"
-	"fmt"
-	"os"
 	"path/filepath"
 	"testing"
 
 	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
-	"github.com/google/uuid"
-	"github.com/stretchr/testify/require"
-	"github.com/stretchr/testify/suite"
+	"github.com/bacalhau-project/bacalhau/pkg/compute/store/test"
 )
 
-type Suite struct {
-	suite.Suite
-	ctx                 context.Context
-	executionStore      store.ExecutionStore
-	localExecutionState store.LocalExecutionState
-	execution           *models.Execution
-	dbFile              string
-}
-
-func (s *Suite) SetupTest() {
-	s.ctx = context.Background()
-
-	dir, _ := os.MkdirTemp("", "bacalhau-executionstore")
-	s.dbFile = filepath.Join(dir, "test.boltdb")
-	s.executionStore, _ = NewStore(s.ctx, s.dbFile)
-	s.execution = mock.ExecutionForJob(mock.Job())
-	s.localExecutionState = *store.NewLocalExecutionState(s.execution, "nodeID-1")
-}
-
-func (s *Suite) TearDownTest() {
-	os.Remove(s.dbFile)
-}
-
-func TestSuite(t *testing.T) {
-	suite.Run(t, new(Suite))
-}
-
-func (s *Suite) TestCreateExecution() {
-	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
-	s.Require().NoError(err)
-
-	// verify the execution was created
-	readExecution, err := s.executionStore.GetExecution(s.ctx, s.execution.ID)
-	s.Require().NoError(err)
-	s.Equal(s.localExecutionState, readExecution)
-
-	// verify a history entry was created
-	history, err := s.executionStore.GetExecutionHistory(s.ctx, s.execution.ID)
-	s.Require().NoError(err)
-	s.Len(history, 1)
-	s.verifyHistory(history[0], readExecution, store.ExecutionStateUndefined, newExecutionComment)
-}
-
-func (s *Suite) TestCreateExecution_AlreadyExists() {
-	err := s.executionStore.CreateExecution(context.Background(), s.localExecutionState)
-	s.Require().NoError(err)
-
-	err = s.executionStore.CreateExecution(context.Background(), s.localExecutionState)
-	s.Error(err)
-}
-
-func (s *Suite) TestCreateExecution_InvalidState() {
-	s.localExecutionState.State = store.ExecutionStateRunning
-	err := s.executionStore.CreateExecution(context.Background(), s.localExecutionState)
-	s.Error(err)
-}
-
-func (s *Suite) TestGetExecution_DoesntExist() {
-	_, err := s.executionStore.GetExecution(context.Background(), uuid.NewString())
-	s.ErrorAs(err, &store.ErrExecutionNotFound{})
-}
-
-func (s *Suite) TestGetExecutions() {
-	ctx := context.Background()
-	err := s.executionStore.CreateExecution(ctx, s.localExecutionState)
-	s.Require().NoError(err)
-
-	readExecutions, err := s.executionStore.GetExecutions(ctx, s.execution.JobID)
-	s.Require().NoError(err)
-	s.Len(readExecutions, 1)
-	s.Equal(s.localExecutionState, readExecutions[0])
-
-	// Create another execution for the same job
-	anotherExecution := mock.ExecutionForJob(s.execution.Job)
-	anotherExecutionState := *store.NewLocalExecutionState(anotherExecution, "nodeID")
-	err = s.executionStore.CreateExecution(ctx, anotherExecutionState)
-	s.Require().NoError(err)
-
-	readExecutions, err = s.executionStore.GetExecutions(ctx, s.execution.JobID)
-	s.Require().NoError(err)
-	s.Len(readExecutions, 2)
-	s.Equal(s.localExecutionState, readExecutions[0])
-	s.Equal(anotherExecutionState, readExecutions[1])
-}
-
-func (s *Suite) TestGetLiveExecutions() {
-	ctx := context.Background()
-
-	localExec := store.NewLocalExecutionState(s.execution, "req1")
-	err := s.executionStore.CreateExecution(ctx, *localExec)
-	s.Require().NoError(err)
-
-	err = s.executionStore.UpdateExecutionState(ctx, store.UpdateExecutionStateRequest{
-		ExecutionID: s.execution.ID,
-		NewState:    store.ExecutionStateRunning,
+func TestStore(t *testing.T) {
+	test.RunStoreSuite(t, func(ctx context.Context, dbPath string) (store.ExecutionStore, error) {
+		dbFile := filepath.Join(dbPath, "test.boltdb")
+		return NewStore(ctx, dbFile)
 	})
-	s.Require().NoError(err)
-
-	execs, err := s.executionStore.GetLiveExecutions(ctx)
-	s.Require().NoError(err)
-	s.Require().Equal(1, len(execs))
-	s.Require().Equal(s.execution.ID, execs[0].Execution.ID)
-}
-
-func (s *Suite) TestFullLiveExecutions() {
-	ctx := context.Background()
-
-	requireLive := func(executionID string) {
-		execs, err := s.executionStore.GetLiveExecutions(ctx)
-		s.Require().NoError(err)
-		s.Require().Equal(1, len(execs))
-		s.Require().Equal(executionID, execs[0].Execution.ID)
-	}
-	requireNotLive := func(executionID string) {
-		execs, err := s.executionStore.GetLiveExecutions(ctx)
-		s.Require().NoError(err)
-		s.Require().Equal([]store.LocalExecutionState{}, execs)
-	}
-
-	localExec := store.NewLocalExecutionState(s.execution, "req1")
-	err := s.executionStore.CreateExecution(ctx, *localExec)
-	s.Require().NoError(err)
-
-	type testdata struct {
-		state       store.LocalExecutionStateType
-		requirement func(string)
-	}
-	testcases := []testdata{
-		{store.ExecutionStateBidAccepted, requireNotLive},
-		{store.ExecutionStateCreated, requireNotLive},
-		{store.ExecutionStateRunning, requireLive},
-		{store.ExecutionStateCancelled, requireNotLive},
-	}
-
-	for _, tc := range testcases {
-		s.T().Run(tc.state.String(), func(t *testing.T) {
-			err = s.executionStore.UpdateExecutionState(ctx, store.UpdateExecutionStateRequest{
-				ExecutionID: s.execution.ID,
-				NewState:    tc.state,
-			})
-			s.Require().NoError(err)
-
-			tc.requirement(s.execution.ID)
-		})
-	}
-
-}
-
-func (s *Suite) TestGetMultipleLiveExecutions() {
-	ctx := context.Background()
-
-	for i := 0; i < 3; i++ {
-		exec := mock.ExecutionForJob(mock.Job())
-		exec.ID = fmt.Sprintf("%d", i+1)
-		localExec := store.NewLocalExecutionState(exec, "req1")
-		err := s.executionStore.CreateExecution(ctx, *localExec)
-		s.Require().NoError(err)
-
-		err = s.executionStore.UpdateExecutionState(ctx, store.UpdateExecutionStateRequest{
-			ExecutionID: exec.ID,
-			NewState:    store.ExecutionStateRunning,
-		})
-		s.Require().NoError(err)
-	}
-
-	execs, err := s.executionStore.GetLiveExecutions(ctx)
-	s.Require().NoError(err)
-	s.Require().Equal(3, len(execs))
-
-	// We want to make sure the executions are returned with increasing update times
-	// that is, oldest first. On windows, the Update times are the same and so we
-	// allow them to compare as <=.
-	s.Require().LessOrEqual(execs[0].UpdateTime, execs[1].UpdateTime)
-	s.Require().LessOrEqual(execs[1].UpdateTime, execs[2].UpdateTime)
-}
-
-func (s *Suite) TestGetExecutions_DoesntExist() {
-	_, err := s.executionStore.GetExecutions(context.Background(), uuid.NewString())
-	s.ErrorAs(err, &store.ErrExecutionsNotFoundForJob{})
-}
-
-func (s *Suite) TestUpdateExecution() {
-	ctx := context.Background()
-	err := s.executionStore.CreateExecution(ctx, s.localExecutionState)
-	s.Require().NoError(err)
-
-	// update with no conditions
-	request := store.UpdateExecutionStateRequest{
-		ExecutionID: s.execution.ID,
-		NewState:    store.ExecutionStatePublishing,
-		Comment:     "Hello There!",
-	}
-	err = s.executionStore.UpdateExecutionState(ctx, request)
-	s.Require().NoError(err)
-
-	// verify the update happened as expected
-	readExecution, err := s.executionStore.GetExecution(ctx, s.execution.ID)
-	s.Require().NoError(err)
-	s.Equal(request.NewState, readExecution.State)
-	s.Equal(s.localExecutionState.Version+1, readExecution.Version)
-
-	// verify a new history entry was created
-	history, err := s.executionStore.GetExecutionHistory(ctx, s.execution.ID)
-	s.Require().NoError(err)
-	s.Len(history, 2)
-	s.verifyHistory(history[1], readExecution, s.localExecutionState.State, request.Comment)
-}
-
-func (s *Suite) TestUpdateExecution_ConditionsPass() {
-	ctx := context.Background()
-	err := s.executionStore.CreateExecution(ctx, s.localExecutionState)
-	s.Require().NoError(err)
-
-	// update with no conditions
-	request := store.UpdateExecutionStateRequest{
-		ExecutionID:     s.execution.ID,
-		ExpectedStates:  []store.LocalExecutionStateType{s.localExecutionState.State},
-		ExpectedVersion: s.localExecutionState.Version,
-		NewState:        store.ExecutionStatePublishing,
-		Comment:         "Hello There!",
-	}
-	err = s.executionStore.UpdateExecutionState(ctx, request)
-	s.Require().NoError(err)
-
-	// verify the update happened as expected
-	readExecution, err := s.executionStore.GetExecution(ctx, s.execution.ID)
-	s.Require().NoError(err)
-	s.Equal(request.NewState, readExecution.State)
-	s.Equal(s.localExecutionState.Version+1, readExecution.Version)
-}
-
-func (s *Suite) TestGetExecutionCount() {
-	ctx := context.Background()
-
-	states := []store.LocalExecutionStateType{
-		store.ExecutionStateBidAccepted,
-		store.ExecutionStateBidAccepted,
-		store.ExecutionStateBidAccepted,
-		store.ExecutionStateCompleted,
-		store.ExecutionStateCompleted,
-	}
-
-	for _, state := range states {
-		execution := mock.ExecutionForJob(mock.Job())
-		executionState := *store.NewLocalExecutionState(execution, "nodeID")
-		err := s.executionStore.CreateExecution(ctx, executionState)
-		s.Require().NoError(err)
-
-		request := store.UpdateExecutionStateRequest{
-			ExecutionID:     execution.ID,
-			ExpectedStates:  []store.LocalExecutionStateType{executionState.State},
-			ExpectedVersion: executionState.Version,
-			NewState:        state,
-			Comment:         "Hello There!",
-		}
-
-		err = s.executionStore.UpdateExecutionState(ctx, request)
-		s.Require().NoError(err)
-	}
-
-	// Close and re-open the execution store so we can
-	// re-populate the counter
-	s.executionStore.Close(s.ctx)
-	s.executionStore, _ = NewStore(s.ctx, s.dbFile)
-
-	c, err := s.executionStore.GetExecutionCount(ctx, store.ExecutionStateCompleted)
-	s.Require().NoError(err)
-	s.Equal(uint64(2), c)
-}
-
-func (s *Suite) TestStateCounterChange() {
-	ctx := context.Background()
-
-	err := s.executionStore.CreateExecution(ctx, s.localExecutionState)
-	s.Require().NoError(err)
-
-	request := store.UpdateExecutionStateRequest{
-		ExecutionID:     s.execution.ID,
-		ExpectedStates:  []store.LocalExecutionStateType{s.localExecutionState.State},
-		ExpectedVersion: s.localExecutionState.Version,
-		NewState:        store.ExecutionStateBidAccepted,
-		Comment:         "Starting",
-	}
-	err = s.executionStore.UpdateExecutionState(ctx, request)
-	s.Require().NoError(err)
-
-	accepted, err := s.executionStore.GetExecutionCount(ctx, store.ExecutionStateBidAccepted)
-	s.Require().NoError(err)
-	completed, err := s.executionStore.GetExecutionCount(ctx, store.ExecutionStateCompleted)
-	s.Require().NoError(err)
-	s.Equal(uint64(1), accepted)
-	s.Equal(uint64(0), completed)
-
-	request = store.UpdateExecutionStateRequest{
-		ExecutionID:     s.execution.ID,
-		ExpectedStates:  []store.LocalExecutionStateType{store.ExecutionStateBidAccepted},
-		ExpectedVersion: s.localExecutionState.Version + 1,
-		NewState:        store.ExecutionStateCompleted,
-		Comment:         "Completed",
-	}
-	err = s.executionStore.UpdateExecutionState(ctx, request)
-	s.Require().NoError(err)
-
-	accepted, err = s.executionStore.GetExecutionCount(ctx, store.ExecutionStateBidAccepted)
-	s.Require().NoError(err)
-	completed, err = s.executionStore.GetExecutionCount(ctx, store.ExecutionStateCompleted)
-	s.Require().NoError(err)
-	s.Equal(uint64(0), accepted)
-	s.Equal(uint64(1), completed)
-
-}
-
-func (s *Suite) TestUpdateExecution_ConditionsStateFail() {
-	ctx := context.Background()
-	err := s.executionStore.CreateExecution(ctx, s.localExecutionState)
-	s.Require().NoError(err)
-
-	// update with no conditions
-	request := store.UpdateExecutionStateRequest{
-		ExecutionID:    s.execution.ID,
-		ExpectedStates: []store.LocalExecutionStateType{store.ExecutionStateBidAccepted},
-		NewState:       store.ExecutionStatePublishing,
-	}
-	err = s.executionStore.UpdateExecutionState(ctx, request)
-	s.ErrorAs(err, &store.ErrInvalidExecutionState{})
-}
-
-func (s *Suite) TestUpdateExecution_ConditionsVersionFail() {
-	ctx := context.Background()
-	err := s.executionStore.CreateExecution(ctx, s.localExecutionState)
-	s.Require().NoError(err)
-
-	// update with no conditions
-	request := store.UpdateExecutionStateRequest{
-		ExecutionID:     s.execution.ID,
-		ExpectedVersion: s.localExecutionState.Version + 99,
-		NewState:        store.ExecutionStatePublishing,
-	}
-	err = s.executionStore.UpdateExecutionState(ctx, request)
-	s.ErrorAs(err, &store.ErrInvalidExecutionVersion{})
-}
-
-func (s *Suite) TestDeleteExecution() {
-	err := s.executionStore.CreateExecution(context.Background(), s.localExecutionState)
-	s.Require().NoError(err)
-
-	err = s.executionStore.DeleteExecution(context.Background(), s.execution.ID)
-	s.Require().NoError(err)
-
-	_, err = s.executionStore.GetExecution(context.Background(), s.execution.ID)
-	s.ErrorAs(err, &store.ErrExecutionNotFound{})
-
-	_, err = s.executionStore.GetExecutions(context.Background(), s.execution.JobID)
-	s.ErrorAs(err, &store.ErrExecutionsNotFoundForJob{})
-}
-
-func (s *Suite) TestDeleteExecution_MultiEntries() {
-	ctx := s.ctx
-	err := s.executionStore.CreateExecution(ctx, s.localExecutionState)
-	s.Require().NoError(err)
-
-	// second execution with same jobID
-	secondExecution := mock.ExecutionForJob(s.execution.Job)
-	secondExecutionState := *store.NewLocalExecutionState(secondExecution, "nodeID")
-	err = s.executionStore.CreateExecution(ctx, secondExecutionState)
-	require.NoError(s.T(), err)
-
-	// third execution with different jobID
-	thirdExecution := mock.ExecutionForJob(mock.Job())
-	thirdExecutionState := *store.NewLocalExecutionState(thirdExecution, "nodeID")
-	err = s.executionStore.CreateExecution(ctx, thirdExecutionState)
-	s.Require().NoError(err)
-
-	// validate pre-state
-	firstJobExecutions, err := s.executionStore.GetExecutions(ctx, s.execution.JobID)
-	s.Require().NoError(err)
-	s.Len(firstJobExecutions, 2)
-
-	secondJobExecutions, err := s.executionStore.GetExecutions(ctx, thirdExecution.JobID)
-	s.Require().NoError(err)
-	s.Len(secondJobExecutions, 1)
-	// delete first execution
-	err = s.executionStore.DeleteExecution(ctx, s.execution.ID)
-	s.Require().NoError(err)
-	_, err = s.executionStore.GetExecution(ctx, s.execution.ID)
-	s.ErrorAs(err, &store.ErrExecutionNotFound{})
-	executions, err := s.executionStore.GetExecutions(ctx, s.execution.JobID)
-	s.Require().NoError(err)
-	s.Len(executions, 1)
-
-	// delete second execution
-	err = s.executionStore.DeleteExecution(ctx, secondExecution.ID)
-	s.Require().NoError(err)
-	_, err = s.executionStore.GetExecution(ctx, secondExecution.ID)
-	s.ErrorAs(err, &store.ErrExecutionNotFound{})
-	_, err = s.executionStore.GetExecutions(ctx, secondExecution.JobID)
-	s.ErrorAs(err, &store.ErrExecutionsNotFoundForJob{})
-
-	// delete third execution
-	err = s.executionStore.DeleteExecution(ctx, thirdExecution.ID)
-	s.Require().NoError(err)
-	_, err = s.executionStore.GetExecution(ctx, thirdExecution.ID)
-	s.ErrorAs(err, &store.ErrExecutionNotFound{})
-	_, err = s.executionStore.GetExecutions(ctx, thirdExecution.JobID)
-	s.ErrorAs(err, &store.ErrExecutionsNotFoundForJob{})
-}
-
-func (s *Suite) TestDeleteExecution_DoesntExist() {
-	err := s.executionStore.DeleteExecution(context.Background(), uuid.NewString())
-	s.ErrorAs(err, &store.ErrExecutionNotFound{})
-}
-
-func (s *Suite) TestGetExecutionHistory_DoesntExist() {
-	_, err := s.executionStore.GetExecutionHistory(context.Background(), uuid.NewString())
-	s.ErrorAs(err, &store.ErrExecutionHistoryNotFound{})
-}
-
-func (s *Suite) verifyHistory(history store.LocalStateHistory, newExecution store.LocalExecutionState, previousState store.LocalExecutionStateType, comment string) {
-	s.Equal(previousState, history.PreviousState)
-	s.Equal(newExecution.Execution.ID, history.ExecutionID)
-	s.Equal(newExecution.State, history.NewState)
-	s.Equal(newExecution.Version, history.NewVersion)
-	s.Equal(newExecution.UpdateTime, history.Time)
-	s.Equal(comment, history.Comment)
 }
diff --git a/pkg/compute/store/test/active_executions_test.go b/pkg/compute/store/test/active_executions_test.go
deleted file mode 100644
index c22d7d8a2e..0000000000
--- a/pkg/compute/store/test/active_executions_test.go
+++ /dev/null
@@ -1,81 +0,0 @@
-//go:build unit || !integration
-
-package test
-
-import (
-	"context"
-	"os"
-	"path/filepath"
-	"testing"
-	"time"
-
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store/boltdb"
-	"github.com/bacalhau-project/bacalhau/pkg/models"
-	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
-	"github.com/google/uuid"
-	"github.com/stretchr/testify/suite"
-)
-
-type Suite struct {
-	suite.Suite
-
-	ctx                 context.Context
-	dbFile              string
-	executionStore      store.ExecutionStore
-	localExecutionState store.LocalExecutionState
-	execution           *models.Execution
-}
-
-func (s *Suite) SetupTest() {
-	s.ctx = context.Background()
-
-	dir, _ := os.MkdirTemp("", "bacalhau-test")
-	s.dbFile = filepath.Join(dir, "test.boltdb")
-
-	s.executionStore, _ = boltdb.NewStore(s.ctx, s.dbFile)
-	s.execution = mock.ExecutionForJob(mock.Job())
-	s.localExecutionState = *store.NewLocalExecutionState(s.execution, "nodeID-1")
-}
-
-func (s *Suite) TearDownTest() {
-	os.Remove(s.dbFile)
-}
-
-func TestSuite(t *testing.T) {
-	suite.Run(t, new(Suite))
-}
-
-func (s *Suite) TestGetActiveExecution_Single() {
-	ctx := context.Background()
-	err := s.executionStore.CreateExecution(ctx, s.localExecutionState)
-	s.NoError(err)
-
-	active, err := store.GetActiveExecution(ctx, s.executionStore, s.execution.JobID)
-	s.NoError(err)
-	s.Equal(s.localExecutionState, active)
-}
-
-func (s *Suite) TestGetActiveExecution_Multiple() {
-	// create a newer execution with same job as the previous one
-	newerExecution := s.execution.Copy()
-	newerExecution.ID = uuid.NewString()
-
-	newerExecutionState := *store.NewLocalExecutionState(newerExecution, "nodeID-1")
-	newerExecutionState.UpdateTime = s.localExecutionState.UpdateTime.Add(time.Second)
-
-	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
-	s.NoError(err)
-
-	err = s.executionStore.CreateExecution(s.ctx, newerExecutionState)
-	s.NoError(err)
-
-	active, err := store.GetActiveExecution(s.ctx, s.executionStore, s.execution.JobID)
-	s.NoError(err)
-	s.Equal(newerExecutionState, active)
-}
-
-func (s *Suite) TestGetActiveExecution_DoestExist() {
-	_, err := store.GetActiveExecution(context.Background(), s.executionStore, s.execution.JobID)
-	s.ErrorAs(err, &store.ErrExecutionsNotFoundForJob{})
-}
diff --git a/pkg/compute/store/test/store_suite.go b/pkg/compute/store/test/store_suite.go
new file mode 100644
index 0000000000..80c3147323
--- /dev/null
+++ b/pkg/compute/store/test/store_suite.go
@@ -0,0 +1,442 @@
+//go:build unit || !integration
+
+package test
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+
+	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
+	"github.com/bacalhau-project/bacalhau/pkg/logger"
+	"github.com/bacalhau-project/bacalhau/pkg/models"
+	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
+)
+
+type StoreCreator func(ctx context.Context, dbpath string) (store.ExecutionStore, error)
+
+type StoreSuite struct {
+	suite.Suite
+	ctx                 context.Context
+	executionStore      store.ExecutionStore
+	localExecutionState store.LocalExecutionState
+	execution           *models.Execution
+	storeCreator        StoreCreator
+	dbPath              string
+}
+
+func (s *StoreSuite) SetupSuite() {
+	logger.ConfigureTestLogging(s.T())
+}
+
+func (s *StoreSuite) SetupTest() {
+	var err error
+	s.ctx = context.Background()
+	s.dbPath = s.T().TempDir()
+	s.executionStore, err = s.storeCreator(s.ctx, s.dbPath)
+	require.NoError(s.T(), err)
+	s.execution = mock.ExecutionForJob(mock.Job())
+	s.localExecutionState = *store.NewLocalExecutionState(s.execution, "nodeID-1")
+}
+
+func (s *StoreSuite) TearDownTest() {
+	if s.executionStore != nil {
+		s.NoError(s.executionStore.Close(s.ctx))
+	}
+	_ = os.Remove(s.dbPath)
+}
+
+func RunStoreSuite(t *testing.T, creator StoreCreator) {
+	s := new(StoreSuite)
+	s.storeCreator = creator
+	suite.Run(t, s)
+}
+
+func (s *StoreSuite) TestCreateExecution() {
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Require().NoError(err)
+
+	// verify the execution was created
+	readExecution, err := s.executionStore.GetExecution(s.ctx, s.execution.ID)
+	s.Require().NoError(err)
+	s.Equal(s.localExecutionState, readExecution)
+
+	// verify a history entry was created
+	history, err := s.executionStore.GetExecutionHistory(s.ctx, s.execution.ID)
+	s.Require().NoError(err)
+	s.Len(history, 1)
+	s.verifyHistory(history[0], readExecution, store.ExecutionStateUndefined, store.NewExecutionMessage)
+}
+
+func (s *StoreSuite) TestCreateExecutionAlreadyExists() {
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Require().NoError(err)
+
+	err = s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Error(err)
+}
+
+func (s *StoreSuite) TestCreateExecutionInvalidState() {
+	s.localExecutionState.State = store.ExecutionStateRunning
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Error(err)
+}
+
+func (s *StoreSuite) TestGetExecutionDoesntExist() {
+	_, err := s.executionStore.GetExecution(s.ctx, uuid.NewString())
+	s.ErrorAs(err, &store.ErrExecutionNotFound{})
+}
+
+func (s *StoreSuite) TestGetExecutions() {
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Require().NoError(err)
+
+	readExecutions, err := s.executionStore.GetExecutions(s.ctx, s.execution.JobID)
+	s.Require().NoError(err)
+	s.Len(readExecutions, 1)
+	s.Equal(s.localExecutionState, readExecutions[0])
+
+	// Create another execution for the same job
+	anotherExecution := mock.ExecutionForJob(s.execution.Job)
+	anotherExecutionState := *store.NewLocalExecutionState(anotherExecution, "nodeID")
+	err = s.executionStore.CreateExecution(s.ctx, anotherExecutionState)
+	s.Require().NoError(err)
+
+	readExecutions, err = s.executionStore.GetExecutions(s.ctx, s.execution.JobID)
+	s.Require().NoError(err)
+	s.Len(readExecutions, 2)
+	s.Equal(s.localExecutionState, readExecutions[0])
+	s.Equal(anotherExecutionState, readExecutions[1])
+}
+
+func (s *StoreSuite) TestGetLiveExecutions() {
+	localExec := store.NewLocalExecutionState(s.execution, "req1")
+	err := s.executionStore.CreateExecution(s.ctx, *localExec)
+	s.Require().NoError(err)
+
+	err = s.executionStore.UpdateExecutionState(s.ctx, store.UpdateExecutionStateRequest{
+		ExecutionID: s.execution.ID,
+		NewState:    store.ExecutionStateRunning,
+	})
+	s.Require().NoError(err)
+
+	execs, err := s.executionStore.GetLiveExecutions(s.ctx)
+	s.Require().NoError(err)
+	s.Require().Equal(1, len(execs))
+	s.Require().Equal(s.execution.ID, execs[0].Execution.ID)
+}
+
+func (s *StoreSuite) TestFullLiveExecutions() {
+	requireLive := func(executionID string) {
+		execs, err := s.executionStore.GetLiveExecutions(s.ctx)
+		s.Require().NoError(err)
+		s.Require().Equal(1, len(execs))
+		s.Require().Equal(executionID, execs[0].Execution.ID)
+	}
+	requireNotLive := func(executionID string) {
+		execs, err := s.executionStore.GetLiveExecutions(s.ctx)
+		s.Require().NoError(err)
+		s.Require().Equal([]store.LocalExecutionState{}, execs)
+	}
+
+	localExec := store.NewLocalExecutionState(s.execution, "req1")
+	err := s.executionStore.CreateExecution(s.ctx, *localExec)
+	s.Require().NoError(err)
+
+	type testdata struct {
+		state       store.LocalExecutionStateType
+		requirement func(string)
+	}
+	testcases := []testdata{
+		{store.ExecutionStateBidAccepted, requireNotLive},
+		{store.ExecutionStateCreated, requireNotLive},
+		{store.ExecutionStateRunning, requireLive},
+		{store.ExecutionStateCancelled, requireNotLive},
+	}
+
+	for _, tc := range testcases {
+		s.T().Run(tc.state.String(), func(t *testing.T) {
+			err = s.executionStore.UpdateExecutionState(s.ctx, store.UpdateExecutionStateRequest{
+				ExecutionID: s.execution.ID,
+				NewState:    tc.state,
+			})
+			s.Require().NoError(err)
+
+			tc.requirement(s.execution.ID)
+		})
+	}
+}
+
+func (s *StoreSuite) TestGetMultipleLiveExecutions() {
+	for i := 0; i < 3; i++ {
+		exec := mock.ExecutionForJob(mock.Job())
+		exec.ID = fmt.Sprintf("%d", i+1)
+		localExec := store.NewLocalExecutionState(exec, "req1")
+		err := s.executionStore.CreateExecution(s.ctx, *localExec)
+		s.Require().NoError(err)
+
+		err = s.executionStore.UpdateExecutionState(s.ctx, store.UpdateExecutionStateRequest{
+			ExecutionID: exec.ID,
+			NewState:    store.ExecutionStateRunning,
+		})
+		s.Require().NoError(err)
+	}
+
+	execs, err := s.executionStore.GetLiveExecutions(s.ctx)
+	s.Require().NoError(err)
+	s.Require().Equal(3, len(execs))
+
+	// We want to make sure the executions are returned with increasing update times
+	// that is, oldest first. On windows, the Update times are the same and so we
+	// allow them to compare as <=.
+	s.Require().LessOrEqual(execs[0].UpdateTime, execs[1].UpdateTime)
+	s.Require().LessOrEqual(execs[1].UpdateTime, execs[2].UpdateTime)
+}
+
+func (s *StoreSuite) TestGetExecutionsDoesntExist() {
+	_, err := s.executionStore.GetExecutions(s.ctx, uuid.NewString())
+	s.ErrorAs(err, &store.ErrExecutionsNotFoundForJob{})
+}
+
+func (s *StoreSuite) TestUpdateExecution() {
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Require().NoError(err)
+
+	// update with no conditions
+	request := store.UpdateExecutionStateRequest{
+		ExecutionID: s.execution.ID,
+		NewState:    store.ExecutionStatePublishing,
+		Comment:     "Hello There!",
+	}
+	err = s.executionStore.UpdateExecutionState(s.ctx, request)
+	s.Require().NoError(err)
+
+	// verify the update happened as expected
+	readExecution, err := s.executionStore.GetExecution(s.ctx, s.execution.ID)
+	s.Require().NoError(err)
+	s.Equal(request.NewState, readExecution.State)
+	s.Equal(s.localExecutionState.Revision+1, readExecution.Revision)
+
+	// verify a new history entry was created
+	history, err := s.executionStore.GetExecutionHistory(s.ctx, s.execution.ID)
+	s.Require().NoError(err)
+	s.Len(history, 2)
+	s.verifyHistory(history[1], readExecution, s.localExecutionState.State, request.Comment)
+}
+
+func (s *StoreSuite) TestUpdateExecutionConditionsPass() {
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Require().NoError(err)
+
+	// update with no conditions
+	request := store.UpdateExecutionStateRequest{
+		ExecutionID:      s.execution.ID,
+		ExpectedStates:   []store.LocalExecutionStateType{s.localExecutionState.State},
+		ExpectedRevision: s.localExecutionState.Revision,
+		NewState:         store.ExecutionStatePublishing,
+		Comment:          "Hello There!",
+	}
+	err = s.executionStore.UpdateExecutionState(s.ctx, request)
+	s.Require().NoError(err)
+
+	// verify the update happened as expected
+	readExecution, err := s.executionStore.GetExecution(s.ctx, s.execution.ID)
+	s.Require().NoError(err)
+	s.Equal(request.NewState, readExecution.State)
+	s.Equal(s.localExecutionState.Revision+1, readExecution.Revision)
+}
+
+func (s *StoreSuite) TestGetExecutionCount() {
+	states := []store.LocalExecutionStateType{
+		store.ExecutionStateBidAccepted,
+		store.ExecutionStateBidAccepted,
+		store.ExecutionStateBidAccepted,
+		store.ExecutionStateCompleted,
+		store.ExecutionStateCompleted,
+	}
+
+	for _, state := range states {
+		execution := mock.ExecutionForJob(mock.Job())
+		executionState := *store.NewLocalExecutionState(execution, "nodeID")
+		err := s.executionStore.CreateExecution(s.ctx, executionState)
+		s.Require().NoError(err)
+
+		request := store.UpdateExecutionStateRequest{
+			ExecutionID:      execution.ID,
+			ExpectedStates:   []store.LocalExecutionStateType{executionState.State},
+			ExpectedRevision: executionState.Revision,
+			NewState:         state,
+			Comment:          "Hello There!",
+		}
+
+		err = s.executionStore.UpdateExecutionState(s.ctx, request)
+		s.Require().NoError(err)
+	}
+
+	// Close and re-open the execution store so we can
+	// re-populate the counter
+	s.Require().NoError(s.executionStore.Close(s.ctx))
+	var err error
+	s.executionStore, err = s.storeCreator(s.ctx, s.dbPath)
+	s.Require().NoError(err)
+
+	c, err := s.executionStore.GetExecutionCount(s.ctx, store.ExecutionStateCompleted)
+	s.Require().NoError(err)
+	s.Equal(uint64(2), c)
+}
+
+func (s *StoreSuite) TestStateCounterChange() {
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Require().NoError(err)
+
+	request := store.UpdateExecutionStateRequest{
+		ExecutionID:      s.execution.ID,
+		ExpectedStates:   []store.LocalExecutionStateType{s.localExecutionState.State},
+		ExpectedRevision: s.localExecutionState.Revision,
+		NewState:         store.ExecutionStateBidAccepted,
+		Comment:          "Starting",
+	}
+	err = s.executionStore.UpdateExecutionState(s.ctx, request)
+	s.Require().NoError(err)
+
+	accepted, err := s.executionStore.GetExecutionCount(s.ctx, store.ExecutionStateBidAccepted)
+	s.Require().NoError(err)
+	completed, err := s.executionStore.GetExecutionCount(s.ctx, store.ExecutionStateCompleted)
+	s.Require().NoError(err)
+	s.Equal(uint64(1), accepted)
+	s.Equal(uint64(0), completed)
+
+	request = store.UpdateExecutionStateRequest{
+		ExecutionID:      s.execution.ID,
+		ExpectedStates:   []store.LocalExecutionStateType{store.ExecutionStateBidAccepted},
+		ExpectedRevision: s.localExecutionState.Revision + 1,
+		NewState:         store.ExecutionStateCompleted,
+		Comment:          "Completed",
+	}
+	err = s.executionStore.UpdateExecutionState(s.ctx, request)
+	s.Require().NoError(err)
+
+	accepted, err = s.executionStore.GetExecutionCount(s.ctx, store.ExecutionStateBidAccepted)
+	s.Require().NoError(err)
+	completed, err = s.executionStore.GetExecutionCount(s.ctx, store.ExecutionStateCompleted)
+	s.Require().NoError(err)
+	s.Equal(uint64(0), accepted)
+	s.Equal(uint64(1), completed)
+}
+
+func (s *StoreSuite) TestUpdateExecutionConditionsStateFail() {
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Require().NoError(err)
+
+	// update with no conditions
+	request := store.UpdateExecutionStateRequest{
+		ExecutionID:    s.execution.ID,
+		ExpectedStates: []store.LocalExecutionStateType{store.ExecutionStateBidAccepted},
+		NewState:       store.ExecutionStatePublishing,
+	}
+	err = s.executionStore.UpdateExecutionState(s.ctx, request)
+	s.ErrorAs(err, &store.ErrInvalidExecutionState{})
+}
+
+func (s *StoreSuite) TestUpdateExecutionConditionsRevisionFail() {
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Require().NoError(err)
+
+	// update with no conditions
+	request := store.UpdateExecutionStateRequest{
+		ExecutionID:      s.execution.ID,
+		ExpectedRevision: s.localExecutionState.Revision + 99,
+		NewState:         store.ExecutionStatePublishing,
+	}
+	err = s.executionStore.UpdateExecutionState(s.ctx, request)
+	s.ErrorAs(err, &store.ErrInvalidExecutionRevision{})
+}
+
+func (s *StoreSuite) TestDeleteExecution() {
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Require().NoError(err)
+
+	err = s.executionStore.DeleteExecution(s.ctx, s.execution.ID)
+	s.Require().NoError(err)
+
+	_, err = s.executionStore.GetExecution(s.ctx, s.execution.ID)
+	s.ErrorAs(err, &store.ErrExecutionNotFound{})
+
+	_, err = s.executionStore.GetExecutions(s.ctx, s.execution.JobID)
+	s.ErrorAs(err, &store.ErrExecutionsNotFoundForJob{})
+}
+
+func (s *StoreSuite) TestDeleteExecutionMultiEntries() {
+	err := s.executionStore.CreateExecution(s.ctx, s.localExecutionState)
+	s.Require().NoError(err)
+
+	// second execution with same jobID
+	secondExecution := mock.ExecutionForJob(s.execution.Job)
+	secondExecutionState := *store.NewLocalExecutionState(secondExecution, "nodeID")
+	err = s.executionStore.CreateExecution(s.ctx, secondExecutionState)
+	require.NoError(s.T(), err)
+
+	// third execution with different jobID
+	thirdExecution := mock.ExecutionForJob(mock.Job())
+	thirdExecutionState := *store.NewLocalExecutionState(thirdExecution, "nodeID")
+	err = s.executionStore.CreateExecution(s.ctx, thirdExecutionState)
+	s.Require().NoError(err)
+
+	// validate pre-state
+	firstJobExecutions, err := s.executionStore.GetExecutions(s.ctx, s.execution.JobID)
+	s.Require().NoError(err)
+	s.Len(firstJobExecutions, 2)
+
+	secondJobExecutions, err := s.executionStore.GetExecutions(s.ctx, thirdExecution.JobID)
+	s.Require().NoError(err)
+	s.Len(secondJobExecutions, 1)
+	// delete first execution
+	err = s.executionStore.DeleteExecution(s.ctx, s.execution.ID)
+	s.Require().NoError(err)
+	_, err = s.executionStore.GetExecution(s.ctx, s.execution.ID)
+	s.ErrorAs(err, &store.ErrExecutionNotFound{})
+	executions, err := s.executionStore.GetExecutions(s.ctx, s.execution.JobID)
+	s.Require().NoError(err)
+	s.Len(executions, 1)
+
+	// delete second execution
+	err = s.executionStore.DeleteExecution(s.ctx, secondExecution.ID)
+	s.Require().NoError(err)
+	_, err = s.executionStore.GetExecution(s.ctx, secondExecution.ID)
+	s.ErrorAs(err, &store.ErrExecutionNotFound{})
+	_, err = s.executionStore.GetExecutions(s.ctx, secondExecution.JobID)
+	s.ErrorAs(err, &store.ErrExecutionsNotFoundForJob{})
+
+	// delete third execution
+	err = s.executionStore.DeleteExecution(s.ctx, thirdExecution.ID)
+	s.Require().NoError(err)
+	_, err = s.executionStore.GetExecution(s.ctx, thirdExecution.ID)
+	s.ErrorAs(err, &store.ErrExecutionNotFound{})
+	_, err = s.executionStore.GetExecutions(s.ctx, thirdExecution.JobID)
+	s.ErrorAs(err, &store.ErrExecutionsNotFoundForJob{})
+}
+
+func (s *StoreSuite) TestDeleteExecutionDoesntExist() {
+	err := s.executionStore.DeleteExecution(s.ctx, uuid.NewString())
+	s.ErrorAs(err, &store.ErrExecutionNotFound{})
+}
+
+func (s *StoreSuite) TestGetExecutionHistoryDoesntExist() {
+	_, err := s.executionStore.GetExecutionHistory(s.ctx, uuid.NewString())
+	s.ErrorAs(err, &store.ErrExecutionHistoryNotFound{})
+}
+
+func (s *StoreSuite) verifyHistory(history store.LocalStateHistory,
+	newExecution store.LocalExecutionState, previousState store.LocalExecutionStateType, comment string) {
+	s.Equal(previousState, history.PreviousState)
+	s.Equal(newExecution.Execution.ID, history.ExecutionID)
+	s.Equal(newExecution.State, history.NewState)
+	s.Equal(newExecution.Revision, history.NewRevision)
+	s.Equal(newExecution.UpdateTime, history.Time)
+	s.Equal(comment, history.Comment)
+}
diff --git a/pkg/compute/store/utils_test.go b/pkg/compute/store/utils_test.go
index f45c464cc4..1b8f12df31 100644
--- a/pkg/compute/store/utils_test.go
+++ b/pkg/compute/store/utils_test.go
@@ -5,8 +5,9 @@ package store
 import (
 	"testing"
 
-	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
 	"github.com/stretchr/testify/assert"
+
+	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
 )
 
 func TestValidateNewExecution(t *testing.T) {
@@ -22,11 +23,11 @@ func TestValidateNewExecution_InvalidState(t *testing.T) {
 	assert.ErrorAs(t, err, &ErrInvalidExecutionState{})
 }
 
-func TestValidateNewExecution_InvalidVersion(t *testing.T) {
+func TestValidateNewExecution_InvalidRevision(t *testing.T) {
 	execution := newExecution()
-	execution.Version = 2
+	execution.Revision = 2
 	err := ValidateNewExecution(execution)
-	assert.ErrorAs(t, err, &ErrInvalidExecutionVersion{})
+	assert.ErrorAs(t, err, &ErrInvalidExecutionRevision{})
 }
 
 func newExecution() LocalExecutionState {