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

Issue:
Remove the compute node's inmemory/inlocal execution stores
This is no longer necessary and so we can remove it.


Discussion / hints:

Fix patch:
diff --git a/docs/docs/dev/databases.md b/docs/docs/dev/databases.md
index 95f5b1b24a..a9d5834bce 100644
--- a/docs/docs/dev/databases.md
+++ b/docs/docs/dev/databases.md
@@ -9,7 +9,7 @@ The location of the database file can be specified using the `BACALHAU_JOB_STORE
 
 ## Compute node database (execution store)
 
-By default, compute nodes store their execution information in an in-memory store (pkg/compute/store/inmemory), but can be configured to persist this information in a boltdb database on disk by setting the `BACALHAU_COMPUTE_STORE_TYPE` environment variable to `boltdb`.
+By default, compute nodes store their execution information in an bolddb-backed store (pkg/compute/store/boltdb).
 
 The location of the database file (for a single node) can be specified using the `BACALHAU_COMPUTE_STORE_PATH` environment variable, which will specify which file to use to store the database.  When not specified, the file will be `{$BACALHAU_DIR}/{NODE_ID}-compute.db`.
 
diff --git a/docs/docs/setting-up/running-node/persistence.md b/docs/docs/setting-up/running-node/persistence.md
index 3d836589f7..33fd4e4efb 100644
--- a/docs/docs/setting-up/running-node/persistence.md
+++ b/docs/docs/setting-up/running-node/persistence.md
@@ -22,7 +22,6 @@ In some cases, it may be preferable to maintain the state in memory, with the ca
 |--|--|--|--|
 |BACALHAU_COMPUTE_STORE_TYPE|--compute-execution-store-type|boltdb|Uses the bolt db execution store (default)|
 |BACALHAU_COMPUTE_STORE_PATH|--compute-execution-store-path|A path (inc. filename)|Specifies where the boltdb database should be stored. Default is `~/.bacalhau/{NODE-ID}-compute/executions.db` if not set|
-|BACALHAU_COMPUTE_STORE_TYPE|--compute-execution-store-type|inmemory|Uses the inmemory execution store|
 
 ## Requester node persistence
 
diff --git a/pkg/compute/store/inlocalstore/store.go b/pkg/compute/store/inlocalstore/store.go
deleted file mode 100644
index f2cd77ef40..0000000000
--- a/pkg/compute/store/inlocalstore/store.go
+++ /dev/null
@@ -1,154 +0,0 @@
-package inlocalstore
-
-import (
-	"context"
-	"encoding/json"
-	"errors"
-	"os"
-	"path/filepath"
-	"sync"
-
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
-	"github.com/bacalhau-project/bacalhau/pkg/storage/util"
-	"github.com/rs/zerolog/log"
-)
-
-type PersistentJobStoreParams struct {
-	Store   store.ExecutionStore
-	RootDir string
-}
-type PersistentExecutionStore struct {
-	store     store.ExecutionStore
-	stateFile string
-	mu        sync.RWMutex
-}
-
-type JobStats struct {
-	JobsCompleted uint64
-}
-
-// Check if the json file exists, and create it if it doesn't.
-func createJobStatsJSONIfNotExists(rootDir string) (string, error) {
-	_, err := os.Stat(rootDir)
-	if err != nil {
-		log.Error().Err(err).Msg("Error reading state root directory")
-		return "", err
-	}
-	jsonFilepath := filepath.Join(rootDir, "jobStats.json")
-	_, err = os.Stat(jsonFilepath)
-	if err != nil {
-		if errors.Is(err, os.ErrNotExist) {
-			log.Debug().Msgf("Creating: %s", jsonFilepath)
-			//Initialise JSON with counter of 0
-			err = writeCounter(jsonFilepath, 0)
-			if err != nil {
-				return "", err
-			}
-		} else {
-			return "", err
-		}
-	}
-	return jsonFilepath, nil
-}
-
-func NewPersistentExecutionStore(params PersistentJobStoreParams) (*PersistentExecutionStore, error) {
-	jsonFilepath, err := createJobStatsJSONIfNotExists(params.RootDir)
-	if err != nil {
-		return nil, err
-	}
-
-	return &PersistentExecutionStore{
-		store:     params.Store,
-		stateFile: jsonFilepath,
-	}, nil
-}
-
-// CreateExecution implements store.ExecutionStore
-func (proxy *PersistentExecutionStore) CreateExecution(ctx context.Context, execution store.LocalExecutionState) error {
-	return proxy.store.CreateExecution(ctx, execution)
-}
-
-// DeleteExecution implements store.ExecutionStore
-func (proxy *PersistentExecutionStore) DeleteExecution(ctx context.Context, id string) error {
-	return proxy.store.DeleteExecution(ctx, id)
-}
-
-// GetExecution implements store.ExecutionStore
-func (proxy *PersistentExecutionStore) GetExecution(ctx context.Context, id string) (store.LocalExecutionState, error) {
-	return proxy.store.GetExecution(ctx, id)
-}
-
-// GetExecutionCount implements store.ExecutionStore
-func (proxy *PersistentExecutionStore) GetExecutionCount(ctx context.Context, _ store.LocalExecutionStateType) (uint64, error) {
-	proxy.mu.RLock()
-	defer proxy.mu.RUnlock()
-	return readCounter(proxy.stateFile)
-}
-
-// GetExecutionHistory implements store.ExecutionStore
-func (proxy *PersistentExecutionStore) GetExecutionHistory(ctx context.Context, id string) ([]store.LocalStateHistory, error) {
-	return proxy.store.GetExecutionHistory(ctx, id)
-}
-
-// GetExecutions implements store.ExecutionStore
-func (proxy *PersistentExecutionStore) GetExecutions(ctx context.Context, sharedID string) ([]store.LocalExecutionState, error) {
-	return proxy.store.GetExecutions(ctx, sharedID)
-}
-
-func (proxy *PersistentExecutionStore) GetLiveExecutions(ctx context.Context) ([]store.LocalExecutionState, error) {
-	return proxy.store.GetLiveExecutions(ctx)
-}
-
-// UpdateExecutionState implements store.ExecutionStore
-func (proxy *PersistentExecutionStore) UpdateExecutionState(ctx context.Context, request store.UpdateExecutionStateRequest) error {
-	err := proxy.store.UpdateExecutionState(ctx, request)
-	if err != nil {
-		return err
-	}
-	if request.NewState == store.ExecutionStateCompleted {
-		proxy.mu.Lock()
-		defer proxy.mu.Unlock()
-		count, err := readCounter(proxy.stateFile)
-		if err != nil {
-			return err
-		}
-		err = writeCounter(proxy.stateFile, count+1)
-		if err != nil {
-			return err
-		}
-	}
-	return err
-}
-
-func (proxy *PersistentExecutionStore) Close(ctx context.Context) error {
-	return proxy.store.Close(ctx)
-}
-
-func writeCounter(filepath string, count uint64) error {
-	var jobStore JobStats
-	jobStore.JobsCompleted += count
-	bs, err := json.Marshal(jobStore)
-	if err != nil {
-		return err
-	}
-	err = os.WriteFile(filepath, bs, util.OS_USER_RW|util.OS_ALL_R)
-	if err != nil {
-		return err
-	}
-	return err
-}
-
-func readCounter(filepath string) (uint64, error) {
-	jsonbs, err := os.ReadFile(filepath)
-	var jobStore JobStats
-	if err != nil {
-		return 0, err
-	}
-	err = json.Unmarshal(jsonbs, &jobStore)
-	if err != nil {
-		return 0, err
-	}
-	return jobStore.JobsCompleted, nil
-}
-
-var _ store.ExecutionStore = (*PersistentExecutionStore)(nil)
diff --git a/pkg/compute/store/inmemory/store.go b/pkg/compute/store/inmemory/store.go
deleted file mode 100644
index 603031d6a2..0000000000
--- a/pkg/compute/store/inmemory/store.go
+++ /dev/null
@@ -1,183 +0,0 @@
-package inmemory
-
-import (
-	"context"
-	"fmt"
-	"sort"
-	"sync"
-	"time"
-
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
-	"golang.org/x/exp/maps"
-)
-
-const newExecutionComment = "LocalExecutionState created"
-
-type Store struct {
-	executionMap map[string]store.LocalExecutionState
-	jobMap       map[string][]string
-	liveMap      map[string]struct{}
-	history      map[string][]store.LocalStateHistory
-	mu           sync.RWMutex
-}
-
-func NewStore() *Store {
-	return &Store{
-		executionMap: make(map[string]store.LocalExecutionState),
-		jobMap:       make(map[string][]string),
-		liveMap:      make(map[string]struct{}),
-		history:      make(map[string][]store.LocalStateHistory),
-	}
-}
-
-func (s *Store) GetExecution(ctx context.Context, id string) (store.LocalExecutionState, error) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	execution, ok := s.executionMap[id]
-	if !ok {
-		return execution, store.NewErrExecutionNotFound(id)
-	}
-	return execution, nil
-}
-
-func (s *Store) GetExecutions(ctx context.Context, jobID string) ([]store.LocalExecutionState, error) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	executionIDs, ok := s.jobMap[jobID]
-	if !ok {
-		return []store.LocalExecutionState{}, store.NewErrExecutionsNotFoundForJob(jobID)
-	}
-	executions := make([]store.LocalExecutionState, len(executionIDs))
-	for i, id := range executionIDs {
-		executions[i] = s.executionMap[id]
-	}
-	return executions, nil
-}
-
-func (s *Store) GetLiveExecutions(ctx context.Context) ([]store.LocalExecutionState, error) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-
-	liveIDs := maps.Keys(s.liveMap)
-
-	executions := make([]store.LocalExecutionState, len(liveIDs))
-	for i, id := range liveIDs {
-		executions[i] = s.executionMap[id]
-	}
-
-	// Ensure executions are returned oldest first
-	sort.Slice(executions, func(i, j int) bool {
-		return executions[i].UpdateTime.Before(executions[j].UpdateTime)
-	})
-
-	return executions, nil
-}
-
-func (s *Store) GetExecutionHistory(ctx context.Context, id string) ([]store.LocalStateHistory, error) {
-	s.mu.RLock()
-	defer s.mu.RUnlock()
-	history, ok := s.history[id]
-	if !ok {
-		return history, store.NewErrExecutionHistoryNotFound(id)
-	}
-	return history, nil
-}
-
-func (s *Store) CreateExecution(ctx context.Context, localExecutionState store.LocalExecutionState) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	execution := localExecutionState.Execution
-	if _, ok := s.executionMap[execution.ID]; ok {
-		return store.NewErrExecutionAlreadyExists(execution.ID)
-	}
-	if err := store.ValidateNewExecution(localExecutionState); err != nil {
-		return fmt.Errorf("CreateExecution failure: %w", err)
-	}
-
-	s.executionMap[execution.ID] = localExecutionState
-	s.jobMap[execution.JobID] = append(s.jobMap[execution.JobID], execution.ID)
-	s.appendHistory(localExecutionState, store.ExecutionStateUndefined, newExecutionComment)
-	return nil
-}
-
-func (s *Store) UpdateExecutionState(ctx context.Context, request store.UpdateExecutionStateRequest) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	localExecutionState, ok := s.executionMap[request.ExecutionID]
-	if !ok {
-		return store.NewErrExecutionNotFound(request.ExecutionID)
-	}
-	if err := request.Validate(localExecutionState); err != nil {
-		return err
-	}
-	if localExecutionState.State.IsTerminal() {
-		return store.NewErrExecutionAlreadyTerminal(request.ExecutionID, localExecutionState.State, request.NewState)
-	}
-
-	previousState := localExecutionState.State
-	localExecutionState.State = request.NewState
-	localExecutionState.Version += 1
-	localExecutionState.UpdateTime = time.Now()
-	s.executionMap[localExecutionState.Execution.ID] = localExecutionState
-	s.appendHistory(localExecutionState, previousState, request.Comment)
-
-	if localExecutionState.State.IsExecuting() {
-		s.liveMap[localExecutionState.Execution.ID] = struct{}{}
-	} else {
-		delete(s.liveMap, localExecutionState.Execution.ID)
-	}
-
-	return nil
-}
-
-func (s *Store) appendHistory(updatedExecution store.LocalExecutionState, previousState store.LocalExecutionStateType, comment string) {
-	historyEntry := store.LocalStateHistory{
-		ExecutionID:   updatedExecution.Execution.ID,
-		PreviousState: previousState,
-		NewState:      updatedExecution.State,
-		NewVersion:    updatedExecution.Version,
-		Comment:       comment,
-		Time:          updatedExecution.UpdateTime,
-	}
-	s.history[updatedExecution.Execution.ID] = append(s.history[updatedExecution.Execution.ID], historyEntry)
-}
-
-func (s *Store) DeleteExecution(ctx context.Context, id string) error {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	localExecutionState, ok := s.executionMap[id]
-	if ok {
-		delete(s.executionMap, id)
-		delete(s.history, id)
-		jobID := localExecutionState.Execution.JobID
-		jobExecutions := s.jobMap[jobID]
-		if len(jobExecutions) == 1 {
-			delete(s.jobMap, jobID)
-		} else {
-			for i, executionID := range jobExecutions {
-				if executionID == id {
-					s.jobMap[jobID] = append(jobExecutions[:i], jobExecutions[i+1:]...)
-					break
-				}
-			}
-		}
-	}
-	return nil
-}
-
-func (s *Store) GetExecutionCount(ctx context.Context, state store.LocalExecutionStateType) (uint64, error) {
-	var counter uint64
-	for _, execution := range s.executionMap {
-		if execution.State == state {
-			counter++
-		}
-	}
-	return counter, nil
-}
-
-func (s *Store) Close(ctx context.Context) error {
-	return nil
-}
-
-// compile-time check that we implement the interface ExecutionStore
-var _ store.ExecutionStore = (*Store)(nil)
diff --git a/pkg/config/types/storagetype.go b/pkg/config/types/storagetype.go
index edd359474c..690a6febe1 100644
--- a/pkg/config/types/storagetype.go
+++ b/pkg/config/types/storagetype.go
@@ -16,8 +16,8 @@ type JobStoreConfig struct {
 type StorageType int64
 
 const (
-	InMemory StorageType = 0
-	BoltDB   StorageType = 1
+	UnknownStorage StorageType = 0
+	BoltDB         StorageType = 1
 )
 
 func (j *StorageType) UnmarshalText(text []byte) error {
@@ -43,13 +43,13 @@ func (j *StorageType) UnmarshalYAML(value *yaml.Node) error {
 }
 
 func ParseStorageType(s string) (ret StorageType, err error) {
-	for typ := InMemory; typ <= BoltDB; typ++ {
+	for typ := UnknownStorage; typ <= BoltDB; typ++ {
 		if equal(typ.String(), s) {
 			return typ, nil
 		}
 	}
 
-	return InMemory, fmt.Errorf("StorageType: unknown type '%s' (valid types: %q)", s, []StorageType{InMemory, BoltDB})
+	return UnknownStorage, fmt.Errorf("StorageType: unknown type '%s' (valid types: %q)", s, []StorageType{BoltDB})
 }
 
 func equal(a, b string) bool {
diff --git a/pkg/config/types/storagetype_string.go b/pkg/config/types/storagetype_string.go
index 9bf8f6bd36..50ad2a704e 100644
--- a/pkg/config/types/storagetype_string.go
+++ b/pkg/config/types/storagetype_string.go
@@ -8,13 +8,13 @@ func _() {
 	// An "invalid array index" compiler error signifies that the constant values have changed.
 	// Re-run the stringer command to generate them again.
 	var x [1]struct{}
-	_ = x[InMemory-0]
+	_ = x[UnknownStorage-0]
 	_ = x[BoltDB-1]
 }
 
-const _StorageType_name = "InMemoryBoltDB"
+const _StorageType_name = "UnknownStorageBoltDB"
 
-var _StorageType_index = [...]uint8{0, 8, 14}
+var _StorageType_index = [...]uint8{0, 14, 20}
 
 func (i StorageType) String() string {
 	if i < 0 || i >= StorageType(len(_StorageType_index)-1) {
diff --git a/pkg/repo/execution_store.go b/pkg/repo/execution_store.go
index 7202f76208..989306447c 100644
--- a/pkg/repo/execution_store.go
+++ b/pkg/repo/execution_store.go
@@ -8,8 +8,6 @@ import (
 
 	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
 	"github.com/bacalhau-project/bacalhau/pkg/compute/store/boltdb"
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store/inlocalstore"
-	memcomputestore "github.com/bacalhau-project/bacalhau/pkg/compute/store/inmemory"
 	"github.com/bacalhau-project/bacalhau/pkg/config"
 	"github.com/bacalhau-project/bacalhau/pkg/config/types"
 )
@@ -33,10 +31,7 @@ func (fsr *FsRepo) InitExecutionStore(ctx context.Context, prefix string) (store
 	if err := config.ForKey(types.NodeComputeExecutionStore, &storeCfg); err != nil {
 		return nil, err
 	}
-	var (
-		store store.ExecutionStore
-		err   error
-	)
+
 	switch storeCfg.Type {
 	case types.BoltDB:
 		path := storeCfg.Path
@@ -44,17 +39,8 @@ func (fsr *FsRepo) InitExecutionStore(ctx context.Context, prefix string) (store
 			path = filepath.Join(stateRootDir, "executions.db")
 		}
 
-		store, err = boltdb.NewStore(ctx, path)
-		if err != nil {
-			return nil, err
-		}
-	case types.InMemory:
-		store = memcomputestore.NewStore()
+		return boltdb.NewStore(ctx, path)
 	default:
 		return nil, fmt.Errorf("unknown JobStore type: %s", storeCfg.Type)
 	}
-	return inlocalstore.NewPersistentExecutionStore(inlocalstore.PersistentJobStoreParams{
-		Store:   store,
-		RootDir: stateRootDir,
-	})
 }


Test patch:
diff --git a/pkg/compute/bidder_test.go b/pkg/compute/bidder_test.go
index a12fc02e59..432654d545 100644
--- a/pkg/compute/bidder_test.go
+++ b/pkg/compute/bidder_test.go
@@ -7,10 +7,11 @@ import (
 	"errors"
 	"fmt"
 	"net/url"
+	"path/filepath"
 	"testing"
 
 	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store/inmemory"
+	"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/stretchr/testify/suite"
@@ -37,10 +38,18 @@ func TestBidderSuite(t *testing.T) {
 }
 
 func (s *BidderSuite) SetupTest() {
+	ctx := context.Background()
+	execStore, err := boltdb.NewStore(ctx,
+		filepath.Join(s.T().TempDir(), "bidder-test.db"))
+	s.Require().NoError(err)
+	s.T().Cleanup(func() {
+		execStore.Close(ctx)
+	})
+
 	s.ctrl = gomock.NewController(s.T())
 	s.mockSemanticStrategy = bidstrategy.NewMockSemanticBidStrategy(s.ctrl)
 	s.mockResourceStrategy = bidstrategy.NewMockResourceBidStrategy(s.ctrl)
-	s.mockExecutionStore = inmemory.NewStore()
+	s.mockExecutionStore = execStore
 	s.mockCallback = compute.NewMockCallback(s.ctrl)
 	s.mockExecutor = compute.NewMockExecutor(s.ctrl)
 	s.bidder = compute.NewBidder(compute.BidderParams{
diff --git a/pkg/compute/startup_test.go b/pkg/compute/startup_test.go
index 566f373027..259bcf891b 100644
--- a/pkg/compute/startup_test.go
+++ b/pkg/compute/startup_test.go
@@ -4,6 +4,7 @@ package compute_test
 
 import (
 	"context"
+	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/suite"
@@ -11,7 +12,7 @@ import (
 
 	"github.com/bacalhau-project/bacalhau/pkg/compute"
 	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store/inmemory"
+	"github.com/bacalhau-project/bacalhau/pkg/compute/store/boltdb"
 	"github.com/bacalhau-project/bacalhau/pkg/models"
 	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
 )
@@ -31,8 +32,12 @@ func (s *StartupTestSuite) SetupTest() {
 
 func (s *StartupTestSuite) TestLongRunning() {
 
-	database := inmemory.NewStore()
-	defer database.Close(s.ctx)
+	ctx := context.Background()
+	database, err := boltdb.NewStore(ctx, filepath.Join(s.T().TempDir(), "bidder-test.db"))
+	s.Require().NoError(err)
+	defer func() {
+		database.Close(ctx)
+	}()
 
 	type testcase struct {
 		ID       string
diff --git a/pkg/compute/store/inlocalstore/store_test.go b/pkg/compute/store/inlocalstore/store_test.go
deleted file mode 100644
index ea0f20da45..0000000000
--- a/pkg/compute/store/inlocalstore/store_test.go
+++ /dev/null
@@ -1,98 +0,0 @@
-//go:build unit || !integration
-
-package inlocalstore
-
-import (
-	"context"
-	"testing"
-
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store"
-	"github.com/bacalhau-project/bacalhau/pkg/compute/store/inmemory"
-	"github.com/bacalhau-project/bacalhau/pkg/test/mock"
-	"github.com/stretchr/testify/suite"
-)
-
-type Suite struct {
-	suite.Suite
-	proxy store.ExecutionStore
-}
-
-func (s *Suite) SetupTest() {
-	var err error
-	s.proxy, err = NewPersistentExecutionStore(PersistentJobStoreParams{
-		Store:   inmemory.NewStore(),
-		RootDir: s.T().TempDir(),
-	})
-	s.Require().NoError(err)
-}
-
-func TestSuite(t *testing.T) {
-	suite.Run(t, new(Suite))
-}
-
-func (s *Suite) TestZeroExecutionsReturnsZeroCount() {
-	count, err := s.proxy.GetExecutionCount(context.Background(), store.ExecutionStateCompleted)
-	s.NoError(err)
-	s.Equal(uint64(0), count)
-
-}
-
-func (s *Suite) TestOneExecutionReturnsOneCount() {
-	//check jobStore is initialised to zero
-	count, err := s.proxy.GetExecutionCount(context.Background(), store.ExecutionStateCompleted)
-	s.NoError(err)
-	s.Equal(uint64(0), count)
-	//create execution
-	execution := mock.ExecutionForJob(mock.Job())
-	executionState := *store.NewLocalExecutionState(execution, "nodeID")
-	err = s.proxy.CreateExecution(context.Background(), executionState)
-	s.NoError(err)
-	err = s.proxy.UpdateExecutionState(context.Background(), store.UpdateExecutionStateRequest{
-		ExecutionID: execution.ID,
-		NewState:    store.ExecutionStateCompleted})
-	s.NoError(err)
-	count, err = s.proxy.GetExecutionCount(context.Background(), store.ExecutionStateCompleted)
-	s.NoError(err)
-	s.Equal(uint64(1), count)
-}
-
-func (s *Suite) TestConsecutiveExecutionsReturnCorrectJobCount() {
-	job := mock.Job()
-	// 3 executions
-	for index := 0; index < 3; index++ {
-		count, err := s.proxy.GetExecutionCount(context.Background(), store.ExecutionStateCompleted)
-		s.NoError(err)
-		s.Equal(uint64(index), count)
-
-		execution := mock.ExecutionForJob(job)
-		executionState := *store.NewLocalExecutionState(execution, "nodeID")
-		err = s.proxy.CreateExecution(context.Background(), executionState)
-		s.NoError(err)
-		err = s.proxy.UpdateExecutionState(context.Background(), store.UpdateExecutionStateRequest{
-			ExecutionID: execution.ID,
-			NewState:    store.ExecutionStateCompleted})
-		s.NoError(err)
-	}
-
-}
-
-func (s *Suite) TestOnlyCompletedJobsIncreaseCounter() {
-	for _, executionStateType := range store.ExecutionStateTypes() {
-		if executionStateType == store.ExecutionStateCompleted {
-			continue
-		}
-		s.Run(executionStateType.String(), func() {
-			execution := mock.ExecutionForJob(mock.Job())
-			executionState := *store.NewLocalExecutionState(execution, "nodeID")
-			err := s.proxy.CreateExecution(context.Background(), executionState)
-			s.NoError(err)
-			err = s.proxy.UpdateExecutionState(context.Background(), store.UpdateExecutionStateRequest{
-				ExecutionID: execution.ID,
-				NewState:    executionStateType})
-			s.NoError(err)
-			count, err := s.proxy.GetExecutionCount(context.Background(), store.ExecutionStateCompleted)
-			s.NoError(err)
-			s.Equal(uint64(0), count)
-		})
-	}
-}
diff --git a/pkg/compute/store/inmemory/store_test.go b/pkg/compute/store/inmemory/store_test.go
deleted file mode 100644
index 062162c506..0000000000
--- a/pkg/compute/store/inmemory/store_test.go
+++ /dev/null
@@ -1,347 +0,0 @@
-//go:build unit || !integration
-
-package inmemory
-
-import (
-	"context"
-	"fmt"
-	"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/suite"
-)
-
-type Suite struct {
-	suite.Suite
-	executionStore      store.ExecutionStore
-	localExecutionState store.LocalExecutionState
-	execution           *models.Execution
-}
-
-func (s *Suite) SetupTest() {
-	s.executionStore = NewStore()
-	s.execution = mock.ExecutionForJob(mock.Job())
-	s.localExecutionState = *store.NewLocalExecutionState(s.execution, "nodeID-1")
-}
-
-func TestSuite(t *testing.T) {
-	suite.Run(t, new(Suite))
-}
-
-func (s *Suite) TestCreateExecution() {
-	err := s.executionStore.CreateExecution(context.Background(), s.localExecutionState)
-	s.NoError(err)
-
-	// verify the execution was created
-	readExecution, err := s.executionStore.GetExecution(context.Background(), s.execution.ID)
-	s.NoError(err)
-	s.Equal(s.localExecutionState, readExecution)
-
-	// verify a history entry was created
-	history, err := s.executionStore.GetExecutionHistory(context.Background(), s.execution.ID)
-	s.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.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.NoError(err)
-
-	readExecutions, err := s.executionStore.GetExecutions(ctx, s.execution.JobID)
-	s.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.NoError(err)
-
-	readExecutions, err = s.executionStore.GetExecutions(ctx, s.execution.JobID)
-	s.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,
-	})
-	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) 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.
-	s.Require().LessOrEqual(execs[0].UpdateTime, execs[1].UpdateTime)
-	s.Require().LessOrEqual(execs[1].UpdateTime, execs[2].UpdateTime)
-}
-
-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) 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.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.NoError(err)
-
-	// verify the update happened as expected
-	readExecution, err := s.executionStore.GetExecution(ctx, s.execution.ID)
-	s.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.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.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.NoError(err)
-
-	// verify the update happened as expected
-	readExecution, err := s.executionStore.GetExecution(ctx, s.execution.ID)
-	s.NoError(err)
-	s.Equal(request.NewState, readExecution.State)
-	s.Equal(s.localExecutionState.Version+1, readExecution.Version)
-}
-
-func (s *Suite) TestUpdateExecution_ConditionsStateFail() {
-	ctx := context.Background()
-	err := s.executionStore.CreateExecution(ctx, s.localExecutionState)
-	s.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.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.NoError(err)
-
-	err = s.executionStore.DeleteExecution(context.Background(), s.execution.ID)
-	s.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 := context.Background()
-	err := s.executionStore.CreateExecution(ctx, s.localExecutionState)
-	s.NoError(err)
-
-	// second execution with same jobID
-	secondExecution := mock.ExecutionForJob(s.execution.Job)
-	secondExecutionState := *store.NewLocalExecutionState(secondExecution, "nodeID")
-	err = s.executionStore.CreateExecution(ctx, secondExecutionState)
-
-	// third execution with different jobID
-	thirdExecution := mock.ExecutionForJob(mock.Job())
-	thirdExecutionState := *store.NewLocalExecutionState(thirdExecution, "nodeID")
-	err = s.executionStore.CreateExecution(ctx, thirdExecutionState)
-	s.NoError(err)
-
-	// validate pre-state
-	firstJobExecutions, err := s.executionStore.GetExecutions(ctx, s.execution.JobID)
-	s.NoError(err)
-	s.Len(firstJobExecutions, 2)
-
-	secondJobExecutions, err := s.executionStore.GetExecutions(ctx, thirdExecution.JobID)
-	s.NoError(err)
-	s.Len(secondJobExecutions, 1)
-
-	// delete first execution
-	err = s.executionStore.DeleteExecution(ctx, s.execution.ID)
-	s.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.NoError(err)
-	s.Len(executions, 1)
-
-	// delete second execution
-	err = s.executionStore.DeleteExecution(ctx, secondExecution.ID)
-	s.NoError(err)
-	_, err = s.executionStore.GetExecution(ctx, secondExecution.ID)
-	s.ErrorAs(err, &store.ErrExecutionNotFound{})
-	executions, err = s.executionStore.GetExecutions(ctx, secondExecution.JobID)
-	s.ErrorAs(err, &store.ErrExecutionsNotFoundForJob{})
-
-	// delete third execution
-	err = s.executionStore.DeleteExecution(ctx, thirdExecution.ID)
-	s.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.NoError(err)
-}
-
-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)
-}