Split DoCreate/DoUpdate from waiting into WaitAfterCreate/Update

Previously DoCreate and DoUpdate blocked on waiter.Wait, preventing
the direct engine from parallelizing operations across resources.
Move the wait into dedicated WaitAfterCreate/WaitAfterUpdate methods
that poll GetSpace until the space reaches SPACE_ACTIVE state.

Addresses review feedback from @andrewnester.

Co-authored-by: Isaac

Author: bernardo-rodriguez

bundle/direct/dresources/app_space.go

@@ -2,10 +2,13 @@ package dresources
 
 import (
 	"context"
+	"fmt"
+	"time"
 
 	"github.com/databricks/cli/bundle/config/resources"
 	"github.com/databricks/databricks-sdk-go"
 	"github.com/databricks/databricks-sdk-go/common/types/fieldmask"
+	"github.com/databricks/databricks-sdk-go/retries"
 	"github.com/databricks/databricks-sdk-go/service/apps"
 )
 
@@ -26,29 +29,53 @@ func (r *ResourceAppSpace) DoRead(ctx context.Context, id string) (*apps.Space,
 }
 
 func (r *ResourceAppSpace) DoCreate(ctx context.Context, config *apps.Space) (string, *apps.Space, error) {
-	waiter, err := r.client.Apps.CreateSpace(ctx, apps.CreateSpaceRequest{
+	// Kick off the create request. Wait for the space to become active in
+	// WaitAfterCreate so that parallel creates are not blocked here.
+	_, err := r.client.Apps.CreateSpace(ctx, apps.CreateSpaceRequest{
 		Space: *config,
 	})
 	if err != nil {
 		return "", nil, err
 	}
-	space, err := waiter.Wait(ctx)
-	if err != nil {
-		return "", nil, err
-	}
-	return space.Name, space, nil
+	return config.Name, nil, nil
 }
 
 func (r *ResourceAppSpace) DoUpdate(ctx context.Context, id string, config *apps.Space, _ *PlanEntry) (*apps.Space, error) {
-	waiter, err := r.client.Apps.UpdateSpace(ctx, apps.UpdateSpaceRequest{
+	_, err := r.client.Apps.UpdateSpace(ctx, apps.UpdateSpaceRequest{
 		Name:       id,
 		Space:      *config,
 		UpdateMask: fieldmask.FieldMask{Paths: []string{"description", "resources", "user_api_scopes", "usage_policy_id"}},
 	})
-	if err != nil {
-		return nil, err
-	}
-	return waiter.Wait(ctx)
+	return nil, err
+}
+
+func (r *ResourceAppSpace) WaitAfterCreate(ctx context.Context, config *apps.Space) (*apps.Space, error) {
+	return r.waitForSpaceActive(ctx, config.Name)
+}
+
+func (r *ResourceAppSpace) WaitAfterUpdate(ctx context.Context, config *apps.Space) (*apps.Space, error) {
+	return r.waitForSpaceActive(ctx, config.Name)
+}
+
+func (r *ResourceAppSpace) waitForSpaceActive(ctx context.Context, name string) (*apps.Space, error) {
+	retrier := retries.New[apps.Space](retries.WithTimeout(20 * time.Minute))
+	return retrier.Run(ctx, func(ctx context.Context) (*apps.Space, error) {
+		space, err := r.client.Apps.GetSpace(ctx, apps.GetSpaceRequest{Name: name})
+		if err != nil {
+			return nil, retries.Halt(err)
+		}
+		if space.Status == nil {
+			return nil, retries.Continues("waiting for status")
+		}
+		switch space.Status.State {
+		case apps.SpaceStatusSpaceStateSpaceActive:
+			return space, nil
+		case apps.SpaceStatusSpaceStateSpaceError:
+			return nil, retries.Halt(fmt.Errorf("space %s is in ERROR state: %s", name, space.Status.Message))
+		default:
+			return nil, retries.Continues(fmt.Sprintf("space state: %s", space.Status.State))
+		}
+	})
 }
 
 func (r *ResourceAppSpace) DoDelete(ctx context.Context, id string) error {