Tentative: bump iterations updated_at on prepare to prevent race condition.

Author: apognu

mocks/scenario_iteration_write_repository.go

@@ -1,6 +1,8 @@
 package mocks
 
 import (
+	"context"
+
 	"github.com/stretchr/testify/mock"
 
 	"github.com/checkmarble/marble-backend/models"
@@ -36,3 +38,7 @@ func (s *ScenarioIterationWriteRepository) DeleteScenarioIteration(exec reposito
 	args := s.Called(exec, scenarioIterationId)
 	return args.Error(0)
 }
+
+func (s *ScenarioIterationWriteRepository) TouchUpdatedAt(ctx context.Context, exec repositories.Executor, iterationId string) error {
+	return s.Called(ctx, exec, iterationId).Error(0)
+}

repositories/scenario_iterations_write.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/checkmarble/marble-backend/models"
 	"github.com/checkmarble/marble-backend/models/ast"
@@ -170,3 +171,18 @@ func (repo *MarbleDbRepository) UpdateScenarioIterationVersion(ctx context.Conte
 	)
 	return err
 }
+
+func (repo *MarbleDbRepository) TouchUpdatedAt(ctx context.Context, exec Executor, iterationId string) error {
+	if err := validateMarbleDbExecutor(exec); err != nil {
+		return err
+	}
+
+	return ExecBuilder(
+		ctx,
+		exec,
+		NewQueryBuilder().
+			Update(dbmodels.TABLE_SCENARIO_ITERATIONS).
+			Set("updated_at", time.Now()).
+			Where(squirrel.Eq{"id": iterationId}),
+	)
+}

repositories/scenarios_read.go

@@ -94,7 +94,7 @@ func (repo *MarbleDbRepository) ListLiveIterationsAndNeighbors(ctx context.Conte
 	}
 
 	liveCte := NewQueryBuilder().
-		Select("scenario_id", "version").
+		Select("scenario_id", "version", "updated_at").
 		From(dbmodels.TABLE_SCENARIO_ITERATIONS + " si").
 		InnerJoin("scenarios s on s.live_scenario_iteration_id = si.id").
 		Where(squirrel.Eq{"si.org_id": orgId}).Prefix("with live as(").Suffix(")")
@@ -113,7 +113,7 @@ func (repo *MarbleDbRepository) ListLiveIterationsAndNeighbors(ctx context.Conte
 		LeftJoin(dbmodels.TABLE_RULES + " AS sir ON sir.scenario_iteration_id = si.id").
 		Where(squirrel.And{
 			squirrel.NotEq{"si.version": nil},
-			squirrel.Expr("si.version >= l.version - 1"),
+			squirrel.Expr("si.version >= l.version - 1 or si.updated_at > l.updated_at"),
 		}).
 		GroupBy("si.id")
 

usecases/scenario_iterations_usecase.go

@@ -53,6 +53,7 @@ type IterationUsecaseRepository interface {
 		exec repositories.Executor,
 		scenarioIterationId string,
 	) error
+	TouchUpdatedAt(ctx context.Context, exec repositories.Executor, iterationId string) error
 
 	UpdateRule(ctx context.Context, exec repositories.Executor, rule models.UpdateRuleInput) error
 }

usecases/scenario_publication_usecase_test.go

@@ -23,6 +23,7 @@ type ScenarioPublicationUsecaseTestSuite struct {
 	scenarioFetcher                *mocks.ScenarioFetcher
 	scenarioPublicationsRepository *mocks.ScenarioPublicationRepository
 	scenarioPublisher              *mocks.ScenarioPublisher
+	scenarioIterationToucher       *mocks.ScenarioIterationWriteRepository
 	exec                           *mocks.Executor
 	transaction                    *mocks.Transaction
 	transactionFactory             *mocks.TransactionFactory
@@ -52,6 +53,7 @@ func (suite *ScenarioPublicationUsecaseTestSuite) SetupTest() {
 	suite.scenarioFetcher = new(mocks.ScenarioFetcher)
 	suite.scenarioPublicationsRepository = new(mocks.ScenarioPublicationRepository)
 	suite.scenarioPublisher = new(mocks.ScenarioPublisher)
+	suite.scenarioIterationToucher = new(mocks.ScenarioIterationWriteRepository)
 	suite.exec = new(mocks.Executor)
 	suite.transaction = new(mocks.Transaction)
 	suite.transactionFactory = &mocks.TransactionFactory{TxMock: suite.transaction}
@@ -159,6 +161,7 @@ func (suite *ScenarioPublicationUsecaseTestSuite) makeUsecase() *ScenarioPublica
 		suite.clientDbIndexEditor,
 		suite.featureAccessReader,
 		nil,
+		suite.scenarioIterationToucher,
 	)
 }
 
@@ -482,6 +485,9 @@ func (suite *ScenarioPublicationUsecaseTestSuite) Test_StartPublicationPreparati
 		[]models.ConcreteIndex{{Indexed: []string{"a", "b"}, Included: []string{"c", "d"}}},
 	).Return(nil)
 
+	suite.executorFactory.On("NewExecutor").Return(suite.transaction)
+	suite.scenarioIterationToucher.On("TouchUpdatedAt", mock.Anything, suite.transaction, suite.iterationId).Return(nil)
+
 	err := suite.makeUsecase().StartPublicationPreparation(suite.ctx, suite.organizationId, suite.iterationId)
 
 	suite.NoError(err)

usecases/scenario_publications_usecase.go

@@ -29,6 +29,10 @@ type ScenarioPublisher interface {
 	) ([]models.ScenarioPublication, error)
 }
 
+type ScenarioPublisherIterationToucher interface {
+	TouchUpdatedAt(ctx context.Context, exec repositories.Executor, iterationId string) error
+}
+
 type clientDbIndexEditor interface {
 	GetIndexesToCreate(ctx context.Context, organizationId string, scenarioIterationId string) (
 		toCreate []models.ConcreteIndex, numPending int, err error,
@@ -69,6 +73,7 @@ type ScenarioPublicationUsecase struct {
 	clientDbIndexEditor            clientDbIndexEditor
 	featureAccessReader            PublicationUsecaseFeatureAccessReader
 	screeningRequirements          ScreeningRequirementChecker
+	scenarioIterationToucher       ScenarioPublisherIterationToucher
 }
 
 func NewScenarioPublicationUsecase(
@@ -82,6 +87,7 @@ func NewScenarioPublicationUsecase(
 	clientDbIndexEditor clientDbIndexEditor,
 	featureAccessReader PublicationUsecaseFeatureAccessReader,
 	screeningRequirements ScreeningRequirementChecker,
+	scenarioIterationToucher ScenarioPublisherIterationToucher,
 ) *ScenarioPublicationUsecase {
 	return &ScenarioPublicationUsecase{
 		transactionFactory:             transactionFactory,
@@ -94,6 +100,7 @@ func NewScenarioPublicationUsecase(
 		clientDbIndexEditor:            clientDbIndexEditor,
 		featureAccessReader:            featureAccessReader,
 		screeningRequirements:          screeningRequirements,
+		scenarioIterationToucher:       scenarioIterationToucher,
 	}
 }
 
@@ -231,6 +238,10 @@ func (usecase *ScenarioPublicationUsecase) StartPublicationPreparation(
 		return models.ErrDataPreparationServiceUnavailable
 	}
 
+	if err := usecase.scenarioIterationToucher.TouchUpdatedAt(ctx, usecase.executorFactory.NewExecutor(), scenarioIterationId); err != nil {
+		return errors.Wrap(err, "could not bump scenario iteration update timestamp")
+	}
+
 	if err := usecase.taskQueueRepository.EnqueueCreateIndexTask(ctx,
 		organizationId, indexesToCreate); err != nil {
 		return err

usecases/usecases_with_creds.go

@@ -257,6 +257,7 @@ func (usecases *UsecasesWithCreds) NewScenarioPublicationUsecase() *ScenarioPubl
 		usecases.NewClientDbIndexEditor(),
 		usecases.NewFeatureAccessReader(),
 		usecases.Repositories.OpenSanctionsRepository,
+		&usecases.Repositories.MarbleDbRepository,
 	)
 }