internal/workflow: add native retry support

Waiting for the workflow to stop before retrying is very annoying. Add
support for retrying while the workflow runs.

Previously, the workflow stopped when no more tasks were runnable. This
raised difficult questions: does the workflow accept retry commands
while it's stopped? Should there be a way to force it to do at least one
task before giving up? Rather than deal with those, change the behavior:
the Run function now returns only on completion or context cancelation.
Update the tests to match. As far as I know, relui itself doesn't care.

Note, though: there is now no way to resume a stopped workflow. I don't
think there's much reason for us to stop them now so I don't think it's
a big problem?

Fixes golang/go#54304.

Change-Id: I4de56a6c50d71dddf0eaafce2e9c135c65e4cfec
Reviewed-on: https://go-review.googlesource.com/c/build/+/422098
Reviewed-by: Jenny Rakoczy <[email protected]>
TryBot-Result: Gopher Robot <[email protected]>
Auto-Submit: Heschi Kreinick <[email protected]>
Run-TryBot: Heschi Kreinick <[email protected]>

Author: heschi

internal/relui/buildrelease_test.go

@@ -300,14 +300,14 @@ func testSecurity(t *testing.T, mergeFixes bool) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	_, err = w.Run(deps.ctx, &verboseListener{t, deps.outputListener})
-	if mergeFixes && err != nil {
-		t.Fatal(err)
-	}
-	if !mergeFixes {
-		if err == nil {
-			t.Fatal("release succeeded without merging fixes to the public repository")
+
+	if mergeFixes {
+		_, err = w.Run(deps.ctx, &verboseListener{t, deps.outputListener})
+		if err != nil {
+			t.Fatal(err)
 		}
+	} else {
+		runToFailure(t, deps.ctx, w, "Check branch state matches source archive", &verboseListener{t, deps.outputListener})
 		return
 	}
 	checkTGZ(t, deps.buildTasks.DownloadURL, deps.publishedFiles, "src.tar.gz", &WebsiteFile{
@@ -818,6 +818,40 @@ func (l *testLogger) Printf(format string, v ...interface{}) {
 	l.t.Logf("task %-10v: LOG: %s", l.task, fmt.Sprintf(format, v...))
 }
 
+func runToFailure(t *testing.T, ctx context.Context, w *workflow.Workflow, task string, wrap workflow.Listener) string {
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	t.Helper()
+	var message string
+	listener := &errorListener{
+		taskName: task,
+		callback: func(m string) {
+			message = m
+			cancel()
+		},
+		Listener: wrap,
+	}
+	_, err := w.Run(ctx, listener)
+	if err == nil {
+		t.Fatalf("workflow unexpectedly succeeded")
+	}
+	return message
+}
+
+type errorListener struct {
+	taskName string
+	callback func(string)
+	workflow.Listener
+}
+
+func (l *errorListener) TaskStateChanged(id uuid.UUID, taskID string, st *workflow.TaskState) error {
+	if st.Name == l.taskName && st.Finished && st.Error != "" {
+		l.callback(st.Error)
+	}
+	l.Listener.TaskStateChanged(id, taskID, st)
+	return nil
+}
+
 // fakeSign acts like a human running the signbinaries job periodically.
 func fakeSign(ctx context.Context, t *testing.T, dir string) {
 	seen := map[string]bool{}

internal/relui/templates/task_list.html

@@ -84,7 +84,7 @@
                     name="task.reset"
                     type="submit"
                     value="Retry"
-                    onclick="return this.form.reportValidity() && confirm('This will retry the task and clear workflow errors.\n\nReady to proceed?')" />
+                    onclick="return this.form.reportValidity() && confirm('This will retry the task.\n\nReady to proceed?')" />
                 </form>
               </div>
             {{else if and (not .ApprovedAt.Valid) (.ReadyForApproval)}}

internal/relui/web.go

@@ -154,7 +154,7 @@ func (s *Server) homeHandler(w http.ResponseWriter, r *http.Request) {
 	}
 	hr := &homeResponse{SiteHeader: s.header}
 	for _, w := range ws {
-		if s.w.running[w.ID.String()] != nil {
+		if _, ok := s.w.running[w.ID.String()]; ok {
 			hr.ActiveWorkflows = append(hr.ActiveWorkflows, w)
 			continue
 		}
@@ -300,43 +300,10 @@ func (s *Server) retryTaskHandler(w http.ResponseWriter, r *http.Request, params
 		http.Error(w, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
 		return
 	}
-	if err := s.retryTask(r.Context(), id, params.ByName("name")); err != nil {
-		log.Printf("s.retryTask(_, %q, %q): %v", id, params.ByName("id"), err)
-		if errors.Is(err, sql.ErrNoRows) || errors.Is(err, pgx.ErrNoRows) {
-			http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
-			return
-		}
-		http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
-		return
+	if err := s.w.RetryTask(r.Context(), id, params.ByName("name")); err != nil {
+		log.Printf("s.w.RetryTask(_, %q): %v", id, err)
 	}
-	if err := s.w.Resume(r.Context(), id); err != nil {
-		log.Printf("s.w.Resume(_, %q): %v", id, err)
-	}
-	http.Redirect(w, r, s.BaseLink("/"), http.StatusSeeOther)
-}
-
-func (s *Server) retryTask(ctx context.Context, id uuid.UUID, name string) error {
-	return s.db.BeginFunc(ctx, func(tx pgx.Tx) error {
-		q := db.New(tx)
-		wf, err := q.Workflow(ctx, id)
-		if err != nil {
-			return fmt.Errorf("q.Workflow: %w", err)
-		}
-		task, err := q.Task(ctx, db.TaskParams{WorkflowID: id, Name: name})
-		if err != nil {
-			return fmt.Errorf("q.Task: %w", err)
-		}
-		if _, err := q.ResetTask(ctx, db.ResetTaskParams{WorkflowID: id, Name: name, UpdatedAt: time.Now()}); err != nil {
-			return fmt.Errorf("q.ResetTask: %w", err)
-		}
-		if _, err := q.ResetWorkflow(ctx, db.ResetWorkflowParams{ID: id, UpdatedAt: time.Now()}); err != nil {
-			return fmt.Errorf("q.ResetWorkflow: %w", err)
-		}
-		l := s.w.l.Logger(id, name)
-		l.Printf("task reset. Previous state: %#v", task)
-		l.Printf("workflow reset. Previous state: %#v", wf)
-		return nil
-	})
+	http.Redirect(w, r, s.BaseLink("/workflows", id.String()), http.StatusSeeOther)
 }
 
 func (s *Server) approveTaskHandler(w http.ResponseWriter, r *http.Request, params httprouter.Params) {

internal/relui/web_test.go

@@ -437,161 +437,6 @@ func TestServerBaseLink(t *testing.T) {
 	}
 }
 
-func TestServerRetryTaskHandler(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	hourAgo := time.Now().Add(-1 * time.Hour)
-	wfID := uuid.New()
-	unchangedWorkflow := db.Workflow{
-		ID:        wfID,
-		Params:    nullString(`{"farewell": "bye", "greeting": "hello"}`),
-		Name:      nullString(`echo`),
-		Finished:  true,
-		Output:    `{"some": "thing"}`,
-		Error:     "internal explosion",
-		CreatedAt: hourAgo, // cmpopts.EquateApproxTime
-		UpdatedAt: hourAgo, // cmpopts.EquateApproxTime
-	}
-
-	cases := []struct {
-		desc          string
-		params        map[string]string
-		wantCode      int
-		wantHeaders   map[string]string
-		wantWorkflows []db.Workflow
-	}{
-		{
-			desc:          "no params",
-			wantCode:      http.StatusNotFound,
-			wantWorkflows: []db.Workflow{unchangedWorkflow},
-		},
-		{
-			desc:          "invalid workflow id",
-			params:        map[string]string{"id": "invalid", "name": "greeting"},
-			wantCode:      http.StatusBadRequest,
-			wantWorkflows: []db.Workflow{unchangedWorkflow},
-		},
-		{
-			desc:          "wrong workflow id",
-			params:        map[string]string{"id": uuid.New().String(), "name": "greeting"},
-			wantCode:      http.StatusNotFound,
-			wantWorkflows: []db.Workflow{unchangedWorkflow},
-		},
-		{
-			desc:     "invalid task name",
-			params:   map[string]string{"id": wfID.String(), "name": "invalid"},
-			wantCode: http.StatusNotFound,
-			wantWorkflows: []db.Workflow{
-				{
-					ID:        wfID,
-					Params:    nullString(`{"farewell": "bye", "greeting": "hello"}`),
-					Name:      nullString(`echo`),
-					Finished:  true,
-					Output:    `{"some": "thing"}`,
-					Error:     "internal explosion",
-					CreatedAt: hourAgo, // cmpopts.EquateApproxTime
-					UpdatedAt: hourAgo, // cmpopts.EquateApproxTime
-				},
-			},
-		},
-		{
-			desc:     "successful reset",
-			params:   map[string]string{"id": wfID.String(), "name": "greeting"},
-			wantCode: http.StatusSeeOther,
-			wantHeaders: map[string]string{
-				"Location": "/",
-			},
-			wantWorkflows: []db.Workflow{
-				{
-					ID:        wfID,
-					Params:    nullString(`{"farewell": "bye", "greeting": "hello"}`),
-					Name:      nullString(`echo`),
-					Output:    "{}",
-					CreatedAt: hourAgo,    // cmpopts.EquateApproxTime
-					UpdatedAt: time.Now(), // cmpopts.EquateApproxTime
-				},
-			},
-		},
-	}
-	for _, c := range cases {
-		t.Run(c.desc, func(t *testing.T) {
-			p := testDB(ctx, t)
-			q := db.New(p)
-
-			wf := db.CreateWorkflowParams{
-				ID:        wfID,
-				Params:    nullString(`{"farewell": "bye", "greeting": "hello"}`),
-				Name:      nullString(`echo`),
-				CreatedAt: hourAgo,
-				UpdatedAt: hourAgo,
-			}
-			if _, err := q.CreateWorkflow(ctx, wf); err != nil {
-				t.Fatalf("CreateWorkflow(_, %v) = _, %v, wanted no error", wf, err)
-			}
-			wff := db.WorkflowFinishedParams{
-				ID:        wf.ID,
-				Finished:  true,
-				Output:    `{"some": "thing"}`,
-				Error:     "internal explosion",
-				UpdatedAt: hourAgo,
-			}
-			if _, err := q.WorkflowFinished(ctx, wff); err != nil {
-				t.Fatalf("WorkflowFinished(_, %v) = _, %v, wanted no error", wff, err)
-			}
-			gtg := db.CreateTaskParams{
-				WorkflowID: wf.ID,
-				Name:       "greeting",
-				Finished:   true,
-				Error:      nullString("internal explosion"),
-				CreatedAt:  hourAgo,
-				UpdatedAt:  hourAgo,
-			}
-			if _, err := q.CreateTask(ctx, gtg); err != nil {
-				t.Fatalf("CreateTask(_, %v) = _, %v, wanted no error", gtg, err)
-			}
-			fw := db.CreateTaskParams{
-				WorkflowID: wf.ID,
-				Name:       "farewell",
-				Finished:   true,
-				Error:      nullString("internal explosion"),
-				CreatedAt:  hourAgo,
-				UpdatedAt:  hourAgo,
-			}
-			if _, err := q.CreateTask(ctx, fw); err != nil {
-				t.Fatalf("CreateTask(_, %v) = _, %v, wanted no error", fw, err)
-			}
-
-			req := httptest.NewRequest(http.MethodPost, path.Join("/workflows/", c.params["id"], "tasks", c.params["name"], "retry"), nil)
-			req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
-			rec := httptest.NewRecorder()
-			s := NewServer(p, NewWorker(NewDefinitionHolder(), p, &PGListener{p}), nil, SiteHeader{}, nil)
-
-			s.m.ServeHTTP(rec, req)
-			resp := rec.Result()
-
-			if resp.StatusCode != c.wantCode {
-				t.Errorf("rep.StatusCode = %d, wanted %d", resp.StatusCode, c.wantCode)
-			}
-			for k, v := range c.wantHeaders {
-				if resp.Header.Get(k) != v {
-					t.Errorf("resp.Header.Get(%q) = %q, wanted %q", k, resp.Header.Get(k), v)
-				}
-			}
-			if c.wantCode == http.StatusBadRequest {
-				return
-			}
-			wfs, err := q.Workflows(ctx)
-			if err != nil {
-				t.Fatalf("q.Workflows() = %v, %v, wanted no error", wfs, err)
-			}
-			if diff := cmp.Diff(c.wantWorkflows, wfs, SameUUIDVariant(), cmpopts.EquateApproxTime(time.Minute)); diff != "" {
-				t.Fatalf("q.Workflows() mismatch (-want +got):\n%s", diff)
-			}
-		})
-	}
-}
-
 func TestServerApproveTaskHandler(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

internal/relui/worker.go

@@ -27,8 +27,6 @@ type Listener interface {
 	WorkflowFinished(ctx context.Context, workflowID uuid.UUID, outputs map[string]interface{}, err error) error
 }
 
-type stopFunc func()
-
 // Worker runs workflows, and persists their state.
 type Worker struct {
 	dh *DefinitionHolder
@@ -43,7 +41,12 @@ type Worker struct {
 	// running is a set of currently running Workflow ids. Run uses
 	// this set to prevent starting a simultaneous execution of a
 	// currently running Workflow.
-	running map[string]stopFunc
+	running map[string]runningWorkflow
+}
+
+type runningWorkflow struct {
+	w    *workflow.Workflow
+	stop func()
 }
 
 // NewWorker returns a Worker ready to accept and run workflows.
@@ -54,7 +57,7 @@ func NewWorker(dh *DefinitionHolder, db db.PGDBTX, l Listener) *Worker {
 		l:       l,
 		done:    make(chan struct{}),
 		pending: make(chan *workflow.Workflow, 1),
-		running: make(map[string]stopFunc),
+		running: make(map[string]runningWorkflow),
 	}
 }
 
@@ -98,7 +101,7 @@ func (w *Worker) markRunning(wf *workflow.Workflow, stop func()) error {
 	if _, ok := w.running[wf.ID.String()]; ok {
 		return fmt.Errorf("workflow %q already running", wf.ID)
 	}
-	w.running[wf.ID.String()] = stop
+	w.running[wf.ID.String()] = runningWorkflow{wf, stop}
 	return nil
 }
 
@@ -111,11 +114,11 @@ func (w *Worker) markStopped(wf *workflow.Workflow) {
 func (w *Worker) cancelWorkflow(id uuid.UUID) bool {
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	stop, ok := w.running[id.String()]
+	rwf, ok := w.running[id.String()]
 	if !ok {
 		return ok
 	}
-	stop()
+	rwf.stop()
 	return ok
 }
 
@@ -236,3 +239,14 @@ func (w *Worker) Resume(ctx context.Context, id uuid.UUID) error {
 	}
 	return w.run(res)
 }
+
+// RetryTask retries a task in a running workflow.
+func (w *Worker) RetryTask(ctx context.Context, id uuid.UUID, name string) error {
+	w.mu.Lock()
+	rwf, ok := w.running[id.String()]
+	w.mu.Unlock()
+	if ok {
+		return fmt.Errorf("no workflow with id %q", id)
+	}
+	return rwf.w.RetryTask(ctx, name)
+}

internal/relui/worker_test.go

@@ -222,13 +222,16 @@ func TestWorkflowResumeRetry(t *testing.T) {
 				counter++
 			}
 		}
+		if counter > 4 {
+			return "", nil
+		}
 		return "", errors.New("expected")
 	})
 	workflow.Output(wd, "nothing", nothing)
 	dh.RegisterDefinition(t.Name(), wd)
 
-	// Run the workflow. It will try the task 3 times and then fail; stop the
-	// worker during its second run, then resume it and verify the task retries.
+	// Run the workflow. It will try the task up to 3 times. Stop the worker
+	// during its second run, then resume it and verify the task retries.
 	go func() {
 		for i := 0; i < 3; i++ {
 			<-blockingChan
@@ -267,9 +270,6 @@ func TestWorkflowResumeRetry(t *testing.T) {
 		t.Fatalf("w.Resume(_, %v) = %v, wanted no error", wfid, err)
 	}
 	<-wfDone
-	if counter-3 != 2 {
-		t.Fatalf("task sent %v more times, wanted 2", counter-3)
-	}
 }
 
 func newTestEchoWorkflow() *workflow.Definition {