fix: set service idempotency logic (#2422)

## Description
Fixes bugs that occurred when running `set_service` consecutive times
against the same enclave.

## Is this change user facing?
NO

core/server/api_container/server/startosis_engine/interpretation_time_value_store/interpretation_time_value_store.go

@@ -83,5 +83,6 @@ func (itvs *InterpretationTimeValueStore) GetNewServiceConfig(name service.Servi
 	if !ok {
 		return nil, stacktrace.NewError("Did not find new service config for '%v' in interpretation time value store.", name)
 	}
+	delete(itvs.setServiceConfigValues, name)
 	return newServiceConfig, nil
 }

core/server/api_container/server/startosis_engine/kurtosis_instruction/set_service/set_service.go

@@ -167,12 +167,19 @@ func (builtin *SetServiceCapabilities) Execute(_ context.Context, _ *builtin_arg
 	return fmt.Sprintf("Set service config on service '%v'.", builtin.serviceName), nil
 }
 
-func (builtin *SetServiceCapabilities) TryResolveWith(instructionsAreEqual bool, _ *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
+func (builtin *SetServiceCapabilities) TryResolveWith(instructionsAreEqual bool, other *enclave_plan_persistence.EnclavePlanInstruction, enclaveComponents *enclave_structure.EnclaveComponents) enclave_structure.InstructionResolutionStatus {
 	if instructionsAreEqual && enclaveComponents.HasServiceBeenUpdated(builtin.serviceName) {
 		return enclave_structure.InstructionIsUpdate
 	} else if instructionsAreEqual {
 		return enclave_structure.InstructionIsEqual
 	}
+
+	// if service names are equal but the instructions are not equal, it means the service config has been updated.
+	// The instruction should be rerun
+	if !instructionsAreEqual {
+		return enclave_structure.InstructionIsUpdate
+	}
+
 	return enclave_structure.InstructionIsUnknown
 }
 

core/server/api_container/server/startosis_engine/startosis_interpreter_idempotent_test.go

@@ -547,7 +547,7 @@ def run(plan):
 	require.False(suite.T(), scheduledInstruction2.IsExecuted()) // set service should also be re-executed because it hasn't been run, but its a noop - its effect is to swap out the service config during interpretation time
 }
 
-func (suite *StartosisInterpreterIdempotentTestSuite) TestStartosisInterpreterIdempotent_MultipleSetService() {
+func (suite *StartosisInterpreterIdempotentTestSuite) TestStartosisInterpreterIdempotent_SetServiceConsecutivelyOnSameService() {
 	initialScript := `
 def run(plan):
 	config = ServiceConfig(
@@ -627,6 +627,115 @@ def run(plan):
 	// TODO: in which case you can `get_service` and assert ports are from the most recent config
 }
 
+func (suite *StartosisInterpreterIdempotentTestSuite) TestStartosisInterpreterIdempotent_SetServiceExecutesConfigUpdateTwice() {
+	initialScript := `
+def run(plan):
+  plan.add_service(name="database", config=ServiceConfig(image="ubuntu"))
+
+  plan.add_service(name="backend", config=ServiceConfig(image="ubuntu"))
+`
+
+	// Interpretation of the initial script to generate the current enclave plan
+	_, currentEnclavePlan, interpretationApiErr := suite.interpreter.Interpret(
+		context.Background(),
+		startosis_constants.PackageIdPlaceholderForStandaloneScript,
+		useDefaultMainFunctionName,
+		noPackageReplaceOptions,
+		startosis_constants.PlaceHolderMainFileForPlaceStandAloneScript,
+		initialScript,
+		noInputParams,
+		defaultNonBlockingMode,
+		enclave_structure.NewEnclaveComponents(),
+		resolver.NewInstructionsPlanMask(0),
+		image_download_mode.ImageDownloadMode_Missing)
+	require.Nil(suite.T(), interpretationApiErr)
+	require.Equal(suite.T(), 2, currentEnclavePlan.Size())
+	convertedEnclavePlan := suite.convertInstructionPlanToEnclavePlan(currentEnclavePlan)
+
+	updatedScript := `
+def run(plan, args):
+  plan.add_service(name="database", config=ServiceConfig(image="ubuntu"))
+
+  plan.add_service(name="backend", config=ServiceConfig(image="ubuntu"))
+
+  plan.set_service(name="database", config=ServiceConfig(image="debian"))
+`
+
+	// Interpret the updated script against the current enclave plan
+	_, instructionsPlan, interpretationError := suite.interpreter.InterpretAndOptimizePlan(
+		context.Background(),
+		startosis_constants.PackageIdPlaceholderForStandaloneScript,
+		noPackageReplaceOptions,
+		useDefaultMainFunctionName,
+		startosis_constants.PlaceHolderMainFileForPlaceStandAloneScript,
+		updatedScript,
+		noInputParams,
+		defaultNonBlockingMode,
+		convertedEnclavePlan,
+		image_download_mode.ImageDownloadMode_Missing,
+	)
+	require.Nil(suite.T(), interpretationError)
+	secondConvertedEnclavePlan := suite.convertInstructionPlanToEnclavePlan(instructionsPlan)
+
+	instructionSequence, err := instructionsPlan.GeneratePlan()
+	require.Nil(suite.T(), err)
+	require.Equal(suite.T(), 0, instructionsPlan.GetIndexOfFirstInstruction())
+	require.Equal(suite.T(), 3, len(instructionSequence))
+
+	scheduledInstruction1 := instructionSequence[0]
+	require.Equal(suite.T(), `add_service(name="database", config=ServiceConfig(image="ubuntu"))`, scheduledInstruction1.GetInstruction().String())
+	require.False(suite.T(), scheduledInstruction1.IsExecuted()) // the add service needs to be re-executed as the `set_service` changed the service config
+
+	scheduledInstruction2 := instructionSequence[1]
+	require.Equal(suite.T(), `add_service(name="backend", config=ServiceConfig(image="ubuntu"))`, scheduledInstruction2.GetInstruction().String())
+	require.True(suite.T(), scheduledInstruction2.IsExecuted()) // this add service should not be re-executed
+
+	scheduledInstruction3 := instructionSequence[2]
+	require.Equal(suite.T(), `set_service(name="database", config=ServiceConfig(image="debian"))`, scheduledInstruction3.GetInstruction().String())
+	require.False(suite.T(), scheduledInstruction3.IsExecuted()) // set service should also be executed because it hasn't been run, but its a noop - its effect is to swap out the service config during interpretation time
+
+	updatedScriptAgain := `
+def run(plan, args):
+  plan.add_service(name="database", config=ServiceConfig(image="ubuntu"))
+
+  plan.add_service(name="backend", config=ServiceConfig(image="ubuntu"))
+
+  plan.set_service(name="database", config=ServiceConfig(image="alpine"))
+`
+
+	// Interpret the second updated script against the current enclave plan
+	_, instructionsPlan, interpretationError = suite.interpreter.InterpretAndOptimizePlan(
+		context.Background(),
+		startosis_constants.PackageIdPlaceholderForStandaloneScript,
+		noPackageReplaceOptions,
+		useDefaultMainFunctionName,
+		startosis_constants.PlaceHolderMainFileForPlaceStandAloneScript,
+		updatedScriptAgain,
+		noInputParams,
+		defaultNonBlockingMode,
+		secondConvertedEnclavePlan,
+		image_download_mode.ImageDownloadMode_Missing,
+	)
+	require.Nil(suite.T(), interpretationError)
+	secondInstructionSequence, err := instructionsPlan.GeneratePlan()
+
+	require.Nil(suite.T(), err)
+	require.Equal(suite.T(), 0, instructionsPlan.GetIndexOfFirstInstruction())
+	require.Equal(suite.T(), 3, len(instructionSequence))
+
+	secondScheduledInstruction1 := secondInstructionSequence[0]
+	require.Equal(suite.T(), `add_service(name="database", config=ServiceConfig(image="ubuntu"))`, secondScheduledInstruction1.GetInstruction().String())
+	require.False(suite.T(), secondScheduledInstruction1.IsExecuted()) // the add service needs to be re-executed as the `set_service` changed the service config
+
+	secondScheduledInstruction2 := secondInstructionSequence[1]
+	require.Equal(suite.T(), `add_service(name="backend", config=ServiceConfig(image="ubuntu"))`, secondScheduledInstruction2.GetInstruction().String())
+	require.True(suite.T(), secondScheduledInstruction2.IsExecuted()) // this add service should not be re-executed
+
+	secondScheduledInstruction3 := secondInstructionSequence[2]
+	require.Equal(suite.T(), `set_service(name="database", config=ServiceConfig(image="alpine"))`, secondScheduledInstruction3.GetInstruction().String())
+	require.False(suite.T(), secondScheduledInstruction3.IsExecuted()) // set service should also be executed because it hasn't been run, but its a noop - its effect is to swap out the service config during interpretation time
+}
+
 func (suite *StartosisInterpreterIdempotentTestSuite) convertInstructionPlanToEnclavePlan(instructionPlan *instructions_plan.InstructionsPlan) *enclave_plan_persistence.EnclavePlan {
 	enclavePlan := enclave_plan_persistence.NewEnclavePlan()
 	instructionPlanSequence, interpretationErr := instructionPlan.GeneratePlan()