cmd/relui,internal/relui: resume workflows

Resume all unfinished workflows on relui startup. Moves most workflow
running logic to a new Worker type.

This also adds a couple new columns to the workflow to record output and
whether a workflow has completed. In order to actually resume workflows,
we'll have to either add functionality to unmark them as "finished"
(restarting manually, which seems reasonable), or not marking a workflow
finished if it terminates with context.Cancelled.

For golang/go#47401

Change-Id: I3e0ed021d7a47fb125f1034df83dc3c6d95887f8
Reviewed-on: https://go-review.googlesource.com/c/build/+/353553
Trust: Alexander Rakoczy <[email protected]>
Run-TryBot: Alexander Rakoczy <[email protected]>
TryBot-Result: Go Bot <[email protected]>
Reviewed-by: Heschi Kreinick <[email protected]>

Author: toothrot

cmd/relui/main.go

@@ -41,7 +41,12 @@ func main() {
 		log.Fatal(err)
 	}
 	defer db.Close()
-	s := relui.NewServer(db)
+	w := relui.NewWorker(db, relui.NewPGListener(db))
+	go w.Run(ctx)
+	if err := w.ResumeAll(ctx); err != nil {
+		log.Printf("w.ResumeAll() = %v", err)
+	}
+	s := relui.NewServer(db, w)
 	if err != nil {
 		log.Fatalf("relui.NewServer() = %v", err)
 	}

cmd/relui/sqlc.yaml

@@ -10,3 +10,6 @@ overrides:
   - go_type: "database/sql.NullString"
     db_type: "jsonb"
     nullable: true
+  - go_type: "string"
+    db_type: "jsonb"
+    nullable: false

internal/relui/db/models.go

@@ -19,15 +19,19 @@ import (
 	"golang.org/x/build/internal/workflow"
 )
 
-// listener implements workflow.Listener for recording workflow state.
-type listener struct {
+func NewPGListener(db *pgxpool.Pool) *PGListener {
+	return &PGListener{db}
+}
+
+// PGListener implements workflow.Listener for recording workflow state.
+type PGListener struct {
 	db *pgxpool.Pool
 }
 
 // TaskStateChanged is called whenever a task is updated by the
 // workflow. The workflow.TaskState is persisted as a db.Task,
 // creating or updating a row as necessary.
-func (l *listener) TaskStateChanged(workflowID uuid.UUID, taskName string, state *workflow.TaskState) error {
+func (l *PGListener) TaskStateChanged(workflowID uuid.UUID, taskName string, state *workflow.TaskState) error {
 	log.Printf("TaskStateChanged(%q, %q, %v)", workflowID, taskName, state)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -55,7 +59,43 @@ func (l *listener) TaskStateChanged(workflowID uuid.UUID, taskName string, state
 	return err
 }
 
-func (l *listener) Logger(workflowID uuid.UUID, taskName string) workflow.Logger {
+// WorkflowStarted persists a new workflow execution in the database.
+func (l *PGListener) WorkflowStarted(ctx context.Context, workflowID uuid.UUID, name string, params map[string]string) error {
+	q := db.New(l.db)
+	m, err := json.Marshal(params)
+	if err != nil {
+		return err
+	}
+	updated := time.Now()
+	_, err = q.CreateWorkflow(ctx, db.CreateWorkflowParams{
+		ID:        workflowID,
+		Name:      sql.NullString{String: name, Valid: true},
+		Params:    sql.NullString{String: string(m), Valid: len(m) > 0},
+		CreatedAt: updated,
+		UpdatedAt: updated,
+	})
+	return err
+}
+
+// WorkflowFinished saves the final state of a workflow after its run
+// has completed.
+func (l *PGListener) WorkflowFinished(ctx context.Context, workflowID uuid.UUID, outputs map[string]interface{}, err error) error {
+	log.Printf("WorkflowCompleted(%q, %v, %q)", workflowID, outputs, err)
+	q := db.New(l.db)
+	m, err := json.Marshal(outputs)
+	if err != nil {
+		return err
+	}
+	_, err = q.WorkflowFinished(ctx, db.WorkflowFinishedParams{
+		ID:        workflowID,
+		Finished:  true,
+		Output:    string(m),
+		UpdatedAt: time.Now(),
+	})
+	return err
+}
+
+func (l *PGListener) Logger(workflowID uuid.UUID, taskName string) workflow.Logger {
 	return &postgresLogger{
 		db:         l.db,
 		workflowID: workflowID,

internal/relui/db/workflows.sql.go

@@ -28,7 +28,7 @@ func TestListenerTaskStateChanged(t *testing.T) {
 		t.Fatalf("q.CreateWorkflow(%v, %v) = %v, wanted no error", ctx, wfp, err)
 	}
 
-	l := &listener{db: dbp}
+	l := &PGListener{db: dbp}
 	state := &workflow.TaskState{
 		Name:             "TestTask",
 		Finished:         true,
@@ -75,7 +75,7 @@ func TestListenerLogger(t *testing.T) {
 		t.Fatalf("q.UpsertTask(%v, %v) = %v, wanted no error", ctx, params, err)
 	}
 
-	l := &listener{db: dbp}
+	l := &PGListener{db: dbp}
 	l.Logger(wf.ID, "TestTask").Printf("A fancy log line says %q", "hello")
 
 	logs, err := q.TaskLogs(ctx)

internal/relui/listener.go

@@ -0,0 +1,10 @@
+-- Copyright 2021 The Go Authors. All rights reserved.
+-- Use of this source code is governed by a BSD-style
+-- license that can be found in the LICENSE file.
+
+BEGIN;
+
+ALTER TABLE workflows DROP COLUMN finished;
+ALTER TABLE workflows DROP COLUMN output;
+
+COMMIT;

internal/relui/listener_test.go

@@ -0,0 +1,26 @@
+-- Copyright 2021 The Go Authors. All rights reserved.
+-- Use of this source code is governed by a BSD-style
+-- license that can be found in the LICENSE file.
+
+BEGIN;
+
+ALTER TABLE workflows
+    ADD COLUMN finished bool NOT NULL DEFAULT false;
+
+CREATE INDEX workflows_finished_ix ON workflows (finished) WHERE finished = false;
+
+ALTER TABLE workflows
+    ADD COLUMN output jsonb NOT NULL DEFAULT jsonb_build_object();
+
+ALTER TABLE workflows
+    ADD COLUMN error text NOT NULL DEFAULT '';
+
+UPDATE workflows
+SET finished = true
+WHERE workflows.id NOT IN (
+    SELECT DISTINCT tasks.workflow_id
+    FROM tasks
+    WHERE finished = false
+);
+
+COMMIT;

internal/relui/migrations/20211005180646_update_workflows_add_finished.down.sql

@@ -3,12 +3,12 @@
 -- license that can be found in the LICENSE file.
 
 -- name: Workflows :many
-SELECT id, params, name, created_at, updated_at
+SELECT *
 FROM workflows
 ORDER BY created_at DESC;
 
 -- name: Workflow :one
-SELECT id, params, name, created_at, updated_at
+SELECT *
 FROM workflows
 WHERE id = $1;
 
@@ -41,7 +41,7 @@ ORDER BY created_at;
 -- name: TasksForWorkflow :many
 SELECT tasks.*
 FROM tasks
-WHERE workflow_id=$1
+WHERE workflow_id = $1
 ORDER BY created_at;
 
 -- name: CreateTaskLog :one
@@ -52,10 +52,25 @@ RETURNING *;
 -- name: TaskLogsForTask :many
 SELECT task_logs.*
 FROM task_logs
-WHERE workflow_id=$1 AND task_name = $2
+WHERE workflow_id = $1
+  AND task_name = $2
 ORDER BY created_at;
 
 -- name: TaskLogs :many
 SELECT task_logs.*
 FROM task_logs
 ORDER BY created_at;
+
+-- name: UnfinishedWorkflows :many
+SELECT workflows.*
+FROM workflows
+WHERE workflows.finished = false;
+
+-- name: WorkflowFinished :one
+UPDATE workflows
+SET finished   = $2,
+    output     = $3,
+    updated_at = $4
+WHERE workflows.id = $1
+RETURNING *;
+