Backend Executor: wait for condition before returning (#14)

* Backend Executor: wait for condition before returning

Signed-off-by: joshvanl <[email protected]>

* Return bare wrapperspb from runtimestate

Signed-off-by: joshvanl <[email protected]>

* Fix runtimestate test

Signed-off-by: joshvanl <[email protected]>

---------

Signed-off-by: joshvanl <[email protected]>

Author: JoshVanL

backend/executor.go

@@ -425,7 +425,10 @@ func (g *grpcExecutor) PurgeInstances(ctx context.Context, req *protos.PurgeInst
 	if err != nil {
 		return resp, fmt.Errorf("failed to purge orchestration state: %w", err)
 	}
-	return resp, nil
+
+	_, err = g.WaitForInstanceCompletion(ctx, &protos.GetInstanceRequest{InstanceId: req.GetInstanceId()})
+
+	return resp, err
 }
 
 // QueryInstances implements protos.TaskHubSidecarServiceServer
@@ -493,7 +496,10 @@ func (g *grpcExecutor) TerminateInstance(ctx context.Context, req *protos.Termin
 	if err := g.backend.AddNewOrchestrationEvent(ctx, api.InstanceID(req.InstanceId), e); err != nil {
 		return nil, fmt.Errorf("failed to submit termination request: %w", err)
 	}
-	return &protos.TerminateResponse{}, nil
+
+	_, err := g.WaitForInstanceCompletion(ctx, &protos.GetInstanceRequest{InstanceId: req.InstanceId})
+
+	return &protos.TerminateResponse{}, err
 }
 
 // SuspendInstance implements protos.TaskHubSidecarServiceServer
@@ -515,7 +521,14 @@ func (g *grpcExecutor) SuspendInstance(ctx context.Context, req *protos.SuspendR
 		return nil, err
 	}
 
-	return &protos.SuspendResponse{}, nil
+	_, err := g.waitForInstance(ctx, &protos.GetInstanceRequest{
+		InstanceId: req.InstanceId,
+	}, func(metadata *OrchestrationMetadata) bool {
+		return metadata.RuntimeStatus == protos.OrchestrationStatus_ORCHESTRATION_STATUS_SUSPENDED ||
+			api.OrchestrationMetadataIsComplete(metadata)
+	})
+
+	return &protos.SuspendResponse{}, err
 }
 
 // ResumeInstance implements protos.TaskHubSidecarServiceServer
@@ -537,7 +550,14 @@ func (g *grpcExecutor) ResumeInstance(ctx context.Context, req *protos.ResumeReq
 		return nil, err
 	}
 
-	return &protos.ResumeResponse{}, nil
+	_, err := g.waitForInstance(ctx, &protos.GetInstanceRequest{
+		InstanceId: req.InstanceId,
+	}, func(metadata *OrchestrationMetadata) bool {
+		return metadata.RuntimeStatus == protos.OrchestrationStatus_ORCHESTRATION_STATUS_RUNNING ||
+			api.OrchestrationMetadataIsComplete(metadata)
+	})
+
+	return &protos.ResumeResponse{}, err
 }
 
 // WaitForInstanceCompletion implements protos.TaskHubSidecarServiceServer

backend/runtimestate/runtimestate.go

@@ -278,22 +278,22 @@ func Name(s *protos.OrchestrationRuntimeState) (string, error) {
 	return s.StartEvent.Name, nil
 }
 
-func Input(s *protos.OrchestrationRuntimeState) (string, error) {
+func Input(s *protos.OrchestrationRuntimeState) (*wrapperspb.StringValue, error) {
 	if s.StartEvent == nil {
-		return "", api.ErrNotStarted
+		return nil, api.ErrNotStarted
 	}
 
 	// REVIEW: Should we distinguish between no input and the empty string?
-	return s.StartEvent.Input.GetValue(), nil
+	return s.StartEvent.Input, nil
 }
 
-func Output(s *protos.OrchestrationRuntimeState) (string, error) {
+func Output(s *protos.OrchestrationRuntimeState) (*wrapperspb.StringValue, error) {
 	if s.CompletedEvent == nil {
-		return "", api.ErrNotCompleted
+		return nil, api.ErrNotCompleted
 	}
 
 	// REVIEW: Should we distinguish between no output and the empty string?
-	return s.CompletedEvent.Result.GetValue(), nil
+	return s.CompletedEvent.Result, nil
 }
 
 func RuntimeStatus(s *protos.OrchestrationRuntimeState) protos.OrchestrationStatus {

tests/runtimestate_test.go

@@ -231,7 +231,7 @@ func Test_RuntimeState_ContinueAsNew(t *testing.T) {
 					assert.Equal(t, expectedName, ec.Name)
 				}
 				if input, err := runtimestate.Input(state); assert.NoError(t, err) {
-					assert.Equal(t, continueAsNewInput, input)
+					assert.Equal(t, continueAsNewInput, input.GetValue())
 				}
 			}
 			assert.NotNil(t, state.NewEvents[2].Timestamp)