[QT-420] Ensure that we exit 1 if we exceed our deadline (#109)

Signed-off-by: Ryan Cragun <[email protected]>

Author: ryancragun

acceptance/scenario_run_test.go

@@ -44,7 +44,7 @@ func TestAcc_Cmd_Scenario_Run(t *testing.T) {
 
 			enos := newAcceptanceRunner(t, skipUnlessTerraformCLI())
 
-			tmpDir, err := os.MkdirTemp("/tmp", "enos.launch")
+			tmpDir, err := os.MkdirTemp("/tmp", "enos.run")
 			require.NoError(t, err)
 			t.Cleanup(func() { os.RemoveAll(tmpDir) })
 
@@ -114,3 +114,43 @@ func TestAcc_Cmd_Scenario_Run(t *testing.T) {
 		})
 	}
 }
+
+// TestAcc_Cmd_Scenario_Run_Timeout tests that a scenario that times out should fail.
+func TestAcc_Cmd_Scenario_Run_Timeout(t *testing.T) {
+	t.Parallel()
+
+	for _, test := range []struct {
+		dir  string
+		name string
+		uid  string
+	}{
+		{
+			"scenario_run_timeout",
+			"timeout",
+			fmt.Sprintf("%x", sha256.Sum256([]byte("timeout"))),
+		},
+	} {
+		test := test
+		t.Run(fmt.Sprintf("%s %s", test.dir, test.name), func(t *testing.T) {
+			t.Parallel()
+
+			enos := newAcceptanceRunner(t, skipUnlessTerraformCLI())
+
+			tmpDir, err := os.MkdirTemp("/tmp", "enos.run")
+			require.NoError(t, err)
+			t.Cleanup(func() { os.RemoveAll(tmpDir) })
+
+			outDir := filepath.Join(tmpDir, test.dir)
+			err = os.MkdirAll(outDir, 0o755)
+			require.NoError(t, err)
+			outDir, err = filepath.EvalSymlinks(outDir)
+			require.NoError(t, err)
+			path, err := filepath.Abs(filepath.Join("./scenarios", test.dir))
+			require.NoError(t, err)
+
+			cmd := fmt.Sprintf("scenario run --chdir %s --out %s --format json --timeout 1s %s", path, outDir, test.name)
+			out, err := enos.run(context.Background(), cmd)
+			require.Error(t, err, string(out))
+		})
+	}
+}

acceptance/scenarios/scenario_run_timeout/enos.hcl

@@ -0,0 +1,9 @@
+module "sleep" {
+  source = "./modules/sleep"
+}
+
+scenario "timeout" {
+  step "sleep" {
+    module = module.sleep
+  }
+}

acceptance/scenarios/scenario_run_timeout/modules/sleep/main.tf

@@ -0,0 +1,3 @@
+resource "time_sleep" "wait_5s" {
+  create_duration = "5s"
+}

internal/client/operation_stream.go

@@ -38,23 +38,104 @@ func (c *Connection) StreamOperations(
 		return res
 	}
 
-	res.Responses = c.streamResponses(ctx, opRes.GetOperations(), ui)
+	var moreDiags []*pb.Diagnostic
+	res.Responses, moreDiags = c.streamResponses(ctx, opRes.GetOperations(), ui)
+	res.Diagnostics = append(res.GetDiagnostics(), moreDiags...)
 
 	return res
 }
 
 // streamResponses takes a context, workspace, and slice of operation references
 // and streams operation events to the ui. It will return a slice of operation
 // responses for each stream that completes.
+//
+//nolint:cyclop // This could probably be refactored to be less complex but right now its inlined.
 func (c *Connection) streamResponses(
 	ctx context.Context,
 	refs []*pb.Ref_Operation,
 	ui uipkg.View,
-) []*pb.Operation_Response {
+) ([]*pb.Operation_Response, []*pb.Diagnostic) {
 	mu := sync.Mutex{}
 	wg := sync.WaitGroup{}
 	res := []*pb.Operation_Response{}
 
+	diags := []*pb.Diagnostic{}
+	diagC := make(chan *pb.Diagnostic)
+	doneC := make(chan struct{})
+	diagWg := sync.WaitGroup{}
+
+	select {
+	case <-ctx.Done():
+		cause := context.Cause(ctx)
+		if cause != nil && cause != context.Canceled {
+			// We have a custom error or our deadline was exceeded.
+			diags = append(diags, diagnostics.FromErr(cause)...)
+		}
+
+		return res, diags
+	default:
+	}
+
+	// Start the error diagnostic routine. This collects diagnostics generated at the client level.
+	// that are unexpected. Per stream request diagnostics will be scoped to each ref.
+	diagWg.Add(1)
+	go func() {
+		defer diagWg.Done()
+
+		drainDiags := func() {
+			for {
+				select {
+				case diag := <-diagC:
+					diags = append(diags, diag)
+
+					continue
+				default:
+				}
+
+				return
+			}
+		}
+
+		checkCtx := func() {
+			err := ctx.Err()
+			if err == nil {
+				return
+			}
+			cause := context.Cause(ctx)
+			if cause != context.Canceled {
+				// We have a custom error or our deadline was exceeded.
+				diags = append(diags, diagnostics.FromErr(cause)...)
+			}
+		}
+
+		for {
+			select {
+			case diag := <-diagC:
+				diags = append(diags, diag)
+
+				continue
+			default:
+			}
+
+			select {
+			case diag := <-diagC:
+				diags = append(diags, diag)
+
+				continue
+			case <-ctx.Done():
+				drainDiags()
+				checkCtx()
+
+				return
+			case <-doneC:
+				drainDiags()
+				checkCtx()
+
+				return
+			}
+		}
+	}()
+
 	for _, ref := range refs {
 		ref := ref
 		wg.Add(1)
@@ -110,6 +191,7 @@ func (c *Connection) streamResponses(
 			for {
 				select {
 				case <-ctx.Done():
+
 					break LOOP
 				case err := <-errC:
 					if err != nil && err != io.EOF {
@@ -168,6 +250,8 @@ func (c *Connection) streamResponses(
 	}
 
 	wg.Wait()
+	close(doneC)
+	diagWg.Wait()
 
-	return res
+	return res, diags
 }

internal/flightplan/matrix.go

@@ -295,6 +295,10 @@ func (v *Vector) EqualUnordered(other *Vector) bool {
 
 // Elements returns a list of the Vectors Elements.
 func (v *Vector) Elements() []Element {
+	if v == nil {
+		return nil
+	}
+
 	return v.elements
 }