Send task summaries as tasks finish

cli/internal/run/real_run.go

@@ -113,6 +113,8 @@ func RealRun(
 			taskSummaries = append(taskSummaries, taskSummary)
 			// not using defer, just release the lock
 			mu.Unlock()
+
+			runSummary.CloseTask(taskSummary)
 		}
 
 		// Return the error when there is one

cli/internal/runsummary/execution_summary.go

@@ -127,6 +127,15 @@ type executionSummary struct {
 	exitCode  int
 }
 
+func (es *executionSummary) Duration() time.Duration {
+	ended := es.endedAt
+	if ended.IsZero() {
+		ended = time.Now()
+	}
+
+	return ended.Sub(es.startedAt)
+}
+
 // MarshalJSON munges the executionSummary into a format we want
 // We'll use an anonmyous, private struct for this, so it's not confusingly duplicated.
 func (es *executionSummary) MarshalJSON() ([]byte, error) {

cli/internal/runsummary/format_execution_summary.go

@@ -17,8 +17,7 @@ func (rsm *Meta) printExecutionSummary() {
 	attempted := summary.ExecutionSummary.attempted
 	successful := summary.ExecutionSummary.cached + summary.ExecutionSummary.success
 	cached := summary.ExecutionSummary.cached
-	// TODO: can we use a method on ExecutionSummary here?
-	duration := time.Since(summary.ExecutionSummary.startedAt).Truncate(time.Millisecond)
+	duration := summary.ExecutionSummary.Duration().Truncate(time.Millisecond)
 
 	if cached == attempted && attempted > 0 {
 		terminalProgram := os.Getenv("TERM_PROGRAM")

cli/internal/runsummary/run_summary.go

@@ -3,10 +3,8 @@ package runsummary
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"path/filepath"
-	"sync"
 	"time"
 
 	"github.com/mitchellh/cli"
@@ -32,9 +30,6 @@ const NoFrameworkDetected = "<NO FRAMEWORK DETECTED>"
 const FrameworkDetectionSkipped = "<FRAMEWORK DETECTION SKIPPED>"
 
 const runSummarySchemaVersion = "0"
-const runsEndpoint = "/v0/spaces/%s/runs"
-const runsPatchEndpoint = "/v0/spaces/%s/runs/%s"
-const tasksEndpoint = "/v0/spaces/%s/runs/%s/tasks"
 
 type runType int
 
@@ -53,8 +48,7 @@ type Meta struct {
 	repoPath           turbopath.RelativeSystemPath
 	singlePackage      bool
 	shouldSave         bool
-	apiClient          *client.APIClient
-	spaceID            string
+	spacesClient       *spacesClient
 	runType            runType
 	synthesizedCommand string
 }
@@ -104,7 +98,7 @@ func NewRunSummary(
 	executionSummary := newExecutionSummary(synthesizedCommand, repoPath, startAt, profile)
 
 	envVars := env.GetEnvMap()
-	return Meta{
+	rsm := Meta{
 		RunSummary: &RunSummary{
 			ID:                 ksuid.New(),
 			Version:            runSummarySchemaVersion,
@@ -123,10 +117,14 @@ func NewRunSummary(
 		repoRoot:           repoRoot,
 		singlePackage:      singlePackage,
 		shouldSave:         shouldSave,
-		apiClient:          apiClient,
-		spaceID:            spaceID,
 		synthesizedCommand: synthesizedCommand,
 	}
+
+	rsm.spacesClient = newSpacesClient(spaceID, apiClient, ui)
+	go rsm.spacesClient.start()
+	rsm.spacesClient.createRun(&rsm)
+
+	return rsm
 }
 
 // getPath returns a path to where the runSummary is written.
@@ -164,47 +162,30 @@ func (rsm *Meta) Close(ctx context.Context, exitCode int, workspaceInfos workspa
 	}
 
 	rsm.printExecutionSummary()
+	rsm.sendToSpace(ctx)
 
-	// If we don't have a spaceID, we can exit now
-	if rsm.spaceID == "" {
-		return nil
-	}
-
-	return rsm.sendToSpace(ctx)
+	return nil
 }
 
-func (rsm *Meta) sendToSpace(ctx context.Context) error {
-	if !rsm.apiClient.IsLinked() {
-		rsm.ui.Warn("Failed to post to space because repo is not linked to a Space. Run `turbo link` first.")
-		return nil
-	}
-
-	// Wrap the record function so we can hoist out url/errors but keep
-	// the function signature/type the spinner.WaitFor expects.
-	var url string
-	var errs []error
-	record := func() {
-		url, errs = rsm.record()
-	}
-
+func (rsm *Meta) sendToSpace(ctx context.Context) {
+	rsm.spacesClient.finishRun(rsm)
 	func() {
-		_ = spinner.WaitFor(ctx, record, rsm.ui, "...sending run summary...", 1000*time.Millisecond)
+		_ = spinner.WaitFor(ctx, rsm.spacesClient.Close, rsm.ui, "...sending run summary...", 1000*time.Millisecond)
 	}()
 
-	// After the spinner is done, print any errors and the url
-	if len(errs) > 0 {
-		rsm.ui.Warn("Errors recording run to Spaces")
-		for _, err := range errs {
-			rsm.ui.Warn(fmt.Sprintf("%v", err))
+	// Print any errors
+	if len(rsm.spacesClient.errors) > 0 {
+		for _, err := range rsm.spacesClient.errors {
+			rsm.ui.Warn(fmt.Sprintf("%s", err))
 		}
 	}
 
+	url := rsm.spacesClient.run.URL
+
 	if url != "" {
 		rsm.ui.Output(fmt.Sprintf("Run: %s", url))
 		rsm.ui.Output("")
 	}
-
-	return nil
 }
 
 // closeDryRun wraps up the Run Summary at the end of `turbo run --dry`.
@@ -248,91 +229,9 @@ func (rsm *Meta) save() error {
 	return summaryPath.WriteFile(json, 0644)
 }
 
-// record sends the summary to the API
-func (rsm *Meta) record() (string, []error) {
-	errs := []error{}
-
-	// Right now we'll send the POST to create the Run and the subsequent task payloads
-	// after all execution is done, but in the future, this first POST request
-	// can happen when the Run actually starts, so we can send updates to the associated Space
-	// as tasks complete.
-	createRunEndpoint := fmt.Sprintf(runsEndpoint, rsm.spaceID)
-	response := &spacesRunResponse{}
-
-	payload := rsm.newSpacesRunCreatePayload()
-	if startPayload, err := json.Marshal(payload); err == nil {
-		if resp, err := rsm.apiClient.JSONPost(createRunEndpoint, startPayload); err != nil {
-			errs = append(errs, fmt.Errorf("POST %s: %w", createRunEndpoint, err))
-		} else {
-			if err := json.Unmarshal(resp, response); err != nil {
-				errs = append(errs, fmt.Errorf("Error unmarshaling response: %w", err))
-			}
-		}
-	}
-
-	if response.ID != "" {
-		if taskErrs := rsm.postTaskSummaries(response.ID); len(taskErrs) > 0 {
-			errs = append(errs, taskErrs...)
-		}
-
-		if donePayload, err := json.Marshal(newSpacesDonePayload(rsm.RunSummary)); err == nil {
-			patchURL := fmt.Sprintf(runsPatchEndpoint, rsm.spaceID, response.ID)
-			if _, err := rsm.apiClient.JSONPatch(patchURL, donePayload); err != nil {
-				errs = append(errs, fmt.Errorf("PATCH %s: %w", patchURL, err))
-			}
-		}
-	}
-
-	if len(errs) > 0 {
-		return response.URL, errs
-	}
-
-	return response.URL, nil
-}
-
-func (rsm *Meta) postTaskSummaries(runID string) []error {
-	errs := []error{}
-	// We make at most 8 requests at a time.
-	maxParallelRequests := 8
-	taskSummaries := rsm.RunSummary.Tasks
-	taskCount := len(taskSummaries)
-	taskURL := fmt.Sprintf(tasksEndpoint, rsm.spaceID, runID)
-
-	parallelRequestCount := maxParallelRequests
-	if taskCount < maxParallelRequests {
-		parallelRequestCount = taskCount
-	}
-
-	queue := make(chan int, taskCount)
-
-	wg := &sync.WaitGroup{}
-	for i := 0; i < parallelRequestCount; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			for index := range queue {
-				task := taskSummaries[index]
-				payload := newSpacesTaskPayload(task)
-				if taskPayload, err := json.Marshal(payload); err == nil {
-					if _, err := rsm.apiClient.JSONPost(taskURL, taskPayload); err != nil {
-						errs = append(errs, fmt.Errorf("Error sending %s summary to space: %w", task.TaskID, err))
-					}
-				}
-			}
-		}()
-	}
-
-	for index := range taskSummaries {
-		queue <- index
-	}
-	close(queue)
-	wg.Wait()
-
-	if len(errs) > 0 {
-		return errs
-	}
-
-	return nil
+// CloseTask posts the result of the Task to Spaces
+func (rsm *Meta) CloseTask(task *TaskSummary) {
+	rsm.spacesClient.postTask(task)
 }
 
 func getUser(envVars env.EnvironmentVariableMap, dir turbopath.AbsoluteSystemPath) string {