feat: Add provisionerdaemon to coderd (#141)

* feat: Add history middleware parameters These will be used for streaming logs, checking status, and other operations related to workspace and project history. * refactor: Move all HTTP routes to top-level struct Nesting all structs behind their respective structures is leaky, and promotes naming conflicts between handlers. Our HTTP routes cannot have conflicts, so neither should function naming. * Add provisioner daemon routes * Add periodic updates * Skip pubsub if short * Return jobs with WorkspaceHistory * Add endpoints for extracting singular history * The full end-to-end operation works * fix: Disable compression for websocket dRPC transport (#145) There is a race condition in the interop between the websocket and `dRPC`: https://github.com/coder/coder/runs/5038545709?check_suite_focus=true#step:7:117 - it seems both the websocket and dRPC feel like they own the `byte[]` being sent between them. This can lead to data races, in which both `dRPC` and the websocket are writing. This is just tracking some experimentation to fix that race condition ## Run results: ## - Run 1: peer test failure - Run 2: peer test failure - Run 3: `TestWorkspaceHistory/CreateHistory` - https://github.com/coder/coder/runs/5040858460?check_suite_focus=true#step:8:45 ``` status code 412: The provided project history is running. Wait for it to complete importing!` ``` - Run 4: `TestWorkspaceHistory/CreateHistory` - https://github.com/coder/coder/runs/5040957999?check_suite_focus=true#step:7:176 ``` workspacehistory_test.go:122: Error Trace: workspacehistory_test.go:122 Error: Condition never satisfied Test: TestWorkspaceHistory/CreateHistory ``` - Run 5: peer failure - Run 6: Pass ✅ - Run 7: Peer failure ## Open Questions: ## ### Is `dRPC` or `websocket` at fault for the data race? It looks like this condition is specifically happening when `dRPC` decides to [`SendError`]). This constructs a new byte payload from [`MarshalError`](https://github.com/storj/drpc/blob/f6e369438f636b47ee788095d3fc13062ffbd019/drpcwire/error.go#L15) - so `dRPC` has created this buffer and owns it. From `dRPC`'s perspective, the callstack looks like this: - [`sendPacket`](https://github.com/storj/drpc/blob/f6e369438f636b47ee788095d3fc13062ffbd019/drpcstream/stream.go#L253) - [`writeFrame`](https://github.com/storj/drpc/blob/f6e369438f636b47ee788095d3fc13062ffbd019/drpcwire/writer.go#L65) - [`AppendFrame`](https://github.com/storj/drpc/blob/f6e369438f636b47ee788095d3fc13062ffbd019/drpcwire/packet.go#L128) - with finally the data race happening here: ```go // AppendFrame appends a marshaled form of the frame to the provided buffer. func AppendFrame(buf []byte, fr Frame) []byte { ... out := buf out = append(out, control). // <--------- ``` This should be fine, since `dPRC` create this buffer, and is taking the byte buffer constructed from `MarshalError` and tacking a bunch of headers on it to create a proper frame. Once `dRPC` is done writing, it _hangs onto the buffer and resets it here__: https://github.com/storj/drpc/blob/f6e369438f636b47ee788095d3fc13062ffbd019/drpcwire/writer.go#L73 However... the websocket implementation, once it gets the buffer, it runs a `statelessDeflate` [here](https://github.com/nhooyr/websocket/blob/8dee580a7f74cf1713400307b4eee514b927870f/write.go#L180), which compresses the buffer on the fly. This functionality actually [mutates the buffer in place](https://github.com/klauspost/compress/blob/a1a9cfc821f00faf2f5231beaa96244344d50391/flate/stateless.go#L94), which is where get our race. In the case where the `byte[]` aren't being manipulated anywhere else, this compress-in-place operation would be safe, and that's probably the case for most over-the-wire usages. In this case, though, where we're plumbing `dRPC` -> websocket, they both are manipulating it (`dRPC` is reusing the buffer for the next `write`, and `websocket` is compressing on the fly). ### Why does cloning on `Read` fail? Get a bunch of errors like: ``` 2022/02/02 19:26:10 [WARN] yamux: frame for missing stream: Vsn:0 Type:0 Flags:0 StreamID:0 Length:0 2022/02/02 19:26:25 [ERR] yamux: Failed to read header: unexpected EOF 2022/02/02 19:26:25 [ERR] yamux: Failed to read header: unexpected EOF 2022/02/02 19:26:25 [WARN] yamux: frame for missing stream: Vsn:0 Type:0 Flags:0 StreamID:0 Length:0 ``` # UPDATE: We decided we could disable websocket compression, which would avoid the race because the in-place `deflate` operaton would no longer be run. Trying that out now: - Run 1: ✅ - Run 2: https://github.com/coder/coder/runs/5042645522?check_suite_focus=true#step:8:338 - Run 3: ✅ - Run 4: https://github.com/coder/coder/runs/5042988758?check_suite_focus=true#step:7:168 - Run 5: ✅ * fix: Remove race condition with acquiredJobDone channel (#148) Found another data race while running the tests: https://github.com/coder/coder/runs/5044320845?check_suite_focus=true#step:7:83 __Issue:__ There is a race in the p.acquiredJobDone chan - in particular, there can be a case where we're waiting on the channel to finish (in close) with <-p.acquiredJobDone, but in parallel, an acquireJob could've been started, which would create a new channel for p.acquiredJobDone. There is a similar race in `close(..)`ing the channel, which also came up in test runs. __Fix:__ Instead of recreating the channel everytime, we can use `sync.WaitGroup` to accomplish the same functionality - a semaphore to make close wait for the current job to wrap up. * fix: Bump up workspace history timeout (#149) This is an attempted fix for failures like: https://github.com/coder/coder/runs/5043435263?check_suite_focus=true#step:7:32 Looking at the timing of the test: ``` t.go:56: 2022-02-02 21:33:21.964 [DEBUG] (terraform-provisioner) <provision.go:139> ran apply t.go:56: 2022-02-02 21:33:21.991 [DEBUG] (provisionerd) <provisionerd.go:162> skipping acquire; job is already running t.go:56: 2022-02-02 21:33:22.050 [DEBUG] (provisionerd) <provisionerd.go:162> skipping acquire; job is already running t.go:56: 2022-02-02 21:33:22.090 [DEBUG] (provisionerd) <provisionerd.go:162> skipping acquire; job is already running t.go:56: 2022-02-02 21:33:22.140 [DEBUG] (provisionerd) <provisionerd.go:162> skipping acquire; job is already running t.go:56: 2022-02-02 21:33:22.195 [DEBUG] (provisionerd) <provisionerd.go:162> skipping acquire; job is already running t.go:56: 2022-02-02 21:33:22.240 [DEBUG] (provisionerd) <provisionerd.go:162> skipping acquire; job is already running workspacehistory_test.go:122: Error Trace: workspacehistory_test.go:122 Error: Condition never satisfied Test: TestWorkspaceHistory/CreateHistory ``` It appears that the `terraform apply` job had just finished - with less than a second to spare until our `require.Eventually` completes - but there's still work to be done (ie, collecting the state files). So my suspicion is that terraform might, in some cases, exceed our 5s timeout. Note that in the setup for this test - there is a similar project history wait that waits for 15s, so I borrowed that here. In the future - we can look at potentially using a simple echo provider to exercise this in the unit test, in a way that is more reliable in terms of timing. I'll log an issue to track that. Co-authored-by: Bryan <bryan@coder.com>
2026-09-24 15:04:27 +08:00 · 2022-02-03 20:34:50 +00:00
parent 35dd208fba
commit e75bde4e31
36 changed files with 1518 additions and 177 deletions
@@ -0,0 +1,619 @@
+package coderd
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"reflect"
+	"time"
+
+	"github.com/go-chi/render"
+	"github.com/google/uuid"
+	"github.com/hashicorp/yamux"
+	"github.com/moby/moby/pkg/namesgenerator"
+	"golang.org/x/xerrors"
+	"nhooyr.io/websocket"
+	"storj.io/drpc/drpcmux"
+	"storj.io/drpc/drpcserver"
+
+	"github.com/coder/coder/coderd/projectparameter"
+	"github.com/coder/coder/database"
+	"github.com/coder/coder/httpapi"
+	"github.com/coder/coder/provisionerd/proto"
+	sdkproto "github.com/coder/coder/provisionersdk/proto"
+)
+
+type ProvisionerDaemon database.ProvisionerDaemon
+
+// Lists all registered provisioner daemons.
+func (api *api) provisionerDaemons(rw http.ResponseWriter, r *http.Request) {
+	daemons, err := api.Database.GetProvisionerDaemons(r.Context())
+	if errors.Is(err, sql.ErrNoRows) {
+		err = nil
+		daemons = []database.ProvisionerDaemon{}
+	}
+	if err != nil {
+		httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+			Message: fmt.Sprintf("get provisioner daemons: %s", err),
+		})
+		return
+	}
+
+	render.Status(r, http.StatusOK)
+	render.JSON(rw, r, daemons)
+}
+
+// Serves the provisioner daemon protobuf API over a WebSocket.
+func (api *api) provisionerDaemonsServe(rw http.ResponseWriter, r *http.Request) {
+	conn, err := websocket.Accept(rw, r, &websocket.AcceptOptions{
+		// Need to disable compression to avoid a data-race
+		CompressionMode: websocket.CompressionDisabled,
+	})
+	if err != nil {
+		httpapi.Write(rw, http.StatusBadRequest, httpapi.Response{
+			Message: fmt.Sprintf("accept websocket: %s", err),
+		})
+		return
+	}
+
+	daemon, err := api.Database.InsertProvisionerDaemon(r.Context(), database.InsertProvisionerDaemonParams{
+		ID:           uuid.New(),
+		CreatedAt:    database.Now(),
+		Name:         namesgenerator.GetRandomName(1),
+		Provisioners: []database.ProvisionerType{database.ProvisionerTypeCdrBasic, database.ProvisionerTypeTerraform},
+	})
+	if err != nil {
+		_ = conn.Close(websocket.StatusInternalError, fmt.Sprintf("insert provisioner daemon:% s", err))
+		return
+	}
+
+	// Multiplexes the incoming connection using yamux.
+	// This allows multiple function calls to occur over
+	// the same connection.
+	session, err := yamux.Server(websocket.NetConn(r.Context(), conn, websocket.MessageBinary), nil)
+	if err != nil {
+		_ = conn.Close(websocket.StatusInternalError, fmt.Sprintf("multiplex server: %s", err))
+		return
+	}
+	mux := drpcmux.New()
+	err = proto.DRPCRegisterProvisionerDaemon(mux, &provisionerdServer{
+		ID:           daemon.ID,
+		Database:     api.Database,
+		Pubsub:       api.Pubsub,
+		Provisioners: daemon.Provisioners,
+	})
+	if err != nil {
+		_ = conn.Close(websocket.StatusInternalError, fmt.Sprintf("drpc register provisioner daemon: %s", err))
+		return
+	}
+	server := drpcserver.New(mux)
+	err = server.Serve(r.Context(), session)
+	if err != nil {
+		_ = conn.Close(websocket.StatusInternalError, fmt.Sprintf("serve: %s", err))
+	}
+}
+
+// The input for a "workspace_provision" job.
+type workspaceProvisionJob struct {
+	WorkspaceHistoryID uuid.UUID `json:"workspace_history_id"`
+}
+
+// The input for a "project_import" job.
+type projectImportJob struct {
+	ProjectHistoryID uuid.UUID `json:"project_history_id"`
+}
+
+// Implementation of the provisioner daemon protobuf server.
+type provisionerdServer struct {
+	ID           uuid.UUID
+	Provisioners []database.ProvisionerType
+	Database     database.Store
+	Pubsub       database.Pubsub
+}
+
+// AcquireJob queries the database to lock a job.
+func (server *provisionerdServer) AcquireJob(ctx context.Context, _ *proto.Empty) (*proto.AcquiredJob, error) {
+	// This marks the job as locked in the database.
+	job, err := server.Database.AcquireProvisionerJob(ctx, database.AcquireProvisionerJobParams{
+		StartedAt: sql.NullTime{
+			Time:  database.Now(),
+			Valid: true,
+		},
+		WorkerID: uuid.NullUUID{
+			UUID:  server.ID,
+			Valid: true,
+		},
+		Types: server.Provisioners,
+	})
+	if errors.Is(err, sql.ErrNoRows) {
+		// The provisioner daemon assumes no jobs are available if
+		// an empty struct is returned.
+		return &proto.AcquiredJob{}, nil
+	}
+	if err != nil {
+		return nil, xerrors.Errorf("acquire job: %w", err)
+	}
+	// Marks the acquired job as failed with the error message provided.
+	failJob := func(errorMessage string) error {
+		err = server.Database.UpdateProvisionerJobByID(ctx, database.UpdateProvisionerJobByIDParams{
+			ID: job.ID,
+			CompletedAt: sql.NullTime{
+				Time:  database.Now(),
+				Valid: true,
+			},
+			Error: sql.NullString{
+				String: errorMessage,
+				Valid:  true,
+			},
+		})
+		if err != nil {
+			return xerrors.Errorf("update provisioner job: %w", err)
+		}
+		return xerrors.Errorf("request job was invalidated: %s", errorMessage)
+	}
+
+	project, err := server.Database.GetProjectByID(ctx, job.ProjectID)
+	if err != nil {
+		return nil, failJob(fmt.Sprintf("get project: %s", err))
+	}
+	organization, err := server.Database.GetOrganizationByID(ctx, project.OrganizationID)
+	if err != nil {
+		return nil, failJob(fmt.Sprintf("get organization: %s", err))
+	}
+	user, err := server.Database.GetUserByID(ctx, job.InitiatorID)
+	if err != nil {
+		return nil, failJob(fmt.Sprintf("get user: %s", err))
+	}
+
+	protoJob := &proto.AcquiredJob{
+		JobId:            job.ID.String(),
+		CreatedAt:        job.CreatedAt.UnixMilli(),
+		Provisioner:      string(job.Provisioner),
+		OrganizationName: organization.Name,
+		ProjectName:      project.Name,
+		UserName:         user.Username,
+	}
+	var projectHistory database.ProjectHistory
+	switch job.Type {
+	case database.ProvisionerJobTypeWorkspaceProvision:
+		var input workspaceProvisionJob
+		err = json.Unmarshal(job.Input, &input)
+		if err != nil {
+			return nil, failJob(fmt.Sprintf("unmarshal job input %q: %s", job.Input, err))
+		}
+		workspaceHistory, err := server.Database.GetWorkspaceHistoryByID(ctx, input.WorkspaceHistoryID)
+		if err != nil {
+			return nil, failJob(fmt.Sprintf("get workspace history: %s", err))
+		}
+		workspace, err := server.Database.GetWorkspaceByID(ctx, workspaceHistory.WorkspaceID)
+		if err != nil {
+			return nil, failJob(fmt.Sprintf("get workspace: %s", err))
+		}
+		projectHistory, err = server.Database.GetProjectHistoryByID(ctx, workspaceHistory.ProjectHistoryID)
+		if err != nil {
+			return nil, failJob(fmt.Sprintf("get project history: %s", err))
+		}
+
+		// Compute parameters for the workspace to consume.
+		parameters, err := projectparameter.Compute(ctx, server.Database, projectparameter.Scope{
+			OrganizationID:     organization.ID,
+			ProjectID:          project.ID,
+			ProjectHistoryID:   projectHistory.ID,
+			UserID:             user.ID,
+			WorkspaceID:        workspace.ID,
+			WorkspaceHistoryID: workspaceHistory.ID,
+		})
+		if err != nil {
+			return nil, failJob(fmt.Sprintf("compute parameters: %s", err))
+		}
+		// Convert parameters to the protobuf type.
+		protoParameters := make([]*sdkproto.ParameterValue, 0, len(parameters))
+		for _, parameter := range parameters {
+			protoParameters = append(protoParameters, parameter.Proto)
+		}
+
+		provisionerState := []byte{}
+		// If workspace history exists before this entry, use that state.
+		// We can't use the before state everytime, because if a job fails
+		// for some random reason, the workspace shouldn't be reset.
+		//
+		// Maybe we should make state global on a workspace?
+		if workspaceHistory.BeforeID.Valid {
+			beforeHistory, err := server.Database.GetWorkspaceHistoryByID(ctx, workspaceHistory.BeforeID.UUID)
+			if err != nil {
+				return nil, failJob(fmt.Sprintf("get workspace history: %s", err))
+			}
+			provisionerState = beforeHistory.ProvisionerState
+		}
+
+		protoJob.Type = &proto.AcquiredJob_WorkspaceProvision_{
+			WorkspaceProvision: &proto.AcquiredJob_WorkspaceProvision{
+				WorkspaceHistoryId: workspaceHistory.ID.String(),
+				WorkspaceName:      workspace.Name,
+				State:              provisionerState,
+				ParameterValues:    protoParameters,
+			},
+		}
+	case database.ProvisionerJobTypeProjectImport:
+		var input projectImportJob
+		err = json.Unmarshal(job.Input, &input)
+		if err != nil {
+			return nil, failJob(fmt.Sprintf("unmarshal job input %q: %s", job.Input, err))
+		}
+		projectHistory, err = server.Database.GetProjectHistoryByID(ctx, input.ProjectHistoryID)
+		if err != nil {
+			return nil, failJob(fmt.Sprintf("get project history: %s", err))
+		}
+
+		protoJob.Type = &proto.AcquiredJob_ProjectImport_{
+			ProjectImport: &proto.AcquiredJob_ProjectImport{
+				ProjectHistoryId:   projectHistory.ID.String(),
+				ProjectHistoryName: projectHistory.Name,
+			},
+		}
+	}
+	switch projectHistory.StorageMethod {
+	case database.ProjectStorageMethodInlineArchive:
+		protoJob.ProjectSourceArchive = projectHistory.StorageSource
+	default:
+		return nil, failJob(fmt.Sprintf("unsupported storage source: %q", projectHistory.StorageMethod))
+	}
+
+	return protoJob, err
+}
+
+func (server *provisionerdServer) UpdateJob(stream proto.DRPCProvisionerDaemon_UpdateJobStream) error {
+	for {
+		update, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		parsedID, err := uuid.Parse(update.JobId)
+		if err != nil {
+			return xerrors.Errorf("parse job id: %w", err)
+		}
+		job, err := server.Database.GetProvisionerJobByID(stream.Context(), parsedID)
+		if err != nil {
+			return xerrors.Errorf("get job: %w", err)
+		}
+		if !job.WorkerID.Valid {
+			return errors.New("job isn't running yet")
+		}
+		if job.WorkerID.UUID.String() != server.ID.String() {
+			return errors.New("you don't own this job")
+		}
+
+		err = server.Database.UpdateProvisionerJobByID(stream.Context(), database.UpdateProvisionerJobByIDParams{
+			ID:        parsedID,
+			UpdatedAt: database.Now(),
+		})
+		if err != nil {
+			return xerrors.Errorf("update job: %w", err)
+		}
+		switch job.Type {
+		case database.ProvisionerJobTypeProjectImport:
+			if len(update.ProjectImportLogs) == 0 {
+				continue
+			}
+			var input projectImportJob
+			err = json.Unmarshal(job.Input, &input)
+			if err != nil {
+				return xerrors.Errorf("unmarshal job input %q: %s", job.Input, err)
+			}
+			insertParams := database.InsertProjectHistoryLogsParams{
+				ProjectHistoryID: input.ProjectHistoryID,
+			}
+			for _, log := range update.ProjectImportLogs {
+				logLevel, err := convertLogLevel(log.Level)
+				if err != nil {
+					return xerrors.Errorf("convert log level: %w", err)
+				}
+				logSource, err := convertLogSource(log.Source)
+				if err != nil {
+					return xerrors.Errorf("convert log source: %w", err)
+				}
+				insertParams.ID = append(insertParams.ID, uuid.New())
+				insertParams.CreatedAt = append(insertParams.CreatedAt, time.UnixMilli(log.CreatedAt))
+				insertParams.Level = append(insertParams.Level, logLevel)
+				insertParams.Source = append(insertParams.Source, logSource)
+				insertParams.Output = append(insertParams.Output, log.Output)
+			}
+			logs, err := server.Database.InsertProjectHistoryLogs(stream.Context(), insertParams)
+			if err != nil {
+				return xerrors.Errorf("insert project logs: %w", err)
+			}
+			data, err := json.Marshal(logs)
+			if err != nil {
+				return xerrors.Errorf("marshal project log: %w", err)
+			}
+			err = server.Pubsub.Publish(projectHistoryLogsChannel(input.ProjectHistoryID), data)
+			if err != nil {
+				return xerrors.Errorf("publish history log: %w", err)
+			}
+		case database.ProvisionerJobTypeWorkspaceProvision:
+			if len(update.WorkspaceProvisionLogs) == 0 {
+				continue
+			}
+			var input workspaceProvisionJob
+			err = json.Unmarshal(job.Input, &input)
+			if err != nil {
+				return xerrors.Errorf("unmarshal job input %q: %s", job.Input, err)
+			}
+			insertParams := database.InsertWorkspaceHistoryLogsParams{
+				WorkspaceHistoryID: input.WorkspaceHistoryID,
+			}
+			for _, log := range update.WorkspaceProvisionLogs {
+				logLevel, err := convertLogLevel(log.Level)
+				if err != nil {
+					return xerrors.Errorf("convert log level: %w", err)
+				}
+				logSource, err := convertLogSource(log.Source)
+				if err != nil {
+					return xerrors.Errorf("convert log source: %w", err)
+				}
+				insertParams.ID = append(insertParams.ID, uuid.New())
+				insertParams.CreatedAt = append(insertParams.CreatedAt, time.UnixMilli(log.CreatedAt))
+				insertParams.Level = append(insertParams.Level, logLevel)
+				insertParams.Source = append(insertParams.Source, logSource)
+				insertParams.Output = append(insertParams.Output, log.Output)
+			}
+			logs, err := server.Database.InsertWorkspaceHistoryLogs(stream.Context(), insertParams)
+			if err != nil {
+				return xerrors.Errorf("insert workspace logs: %w", err)
+			}
+			data, err := json.Marshal(logs)
+			if err != nil {
+				return xerrors.Errorf("marshal project log: %w", err)
+			}
+			err = server.Pubsub.Publish(workspaceHistoryLogsChannel(input.WorkspaceHistoryID), data)
+			if err != nil {
+				return xerrors.Errorf("publish history log: %w", err)
+			}
+		}
+	}
+}
+
+func (server *provisionerdServer) CancelJob(ctx context.Context, cancelJob *proto.CancelledJob) (*proto.Empty, error) {
+	jobID, err := uuid.Parse(cancelJob.JobId)
+	if err != nil {
+		return nil, xerrors.Errorf("parse job id: %w", err)
+	}
+	err = server.Database.UpdateProvisionerJobByID(ctx, database.UpdateProvisionerJobByIDParams{
+		ID: jobID,
+		CancelledAt: sql.NullTime{
+			Time:  database.Now(),
+			Valid: true,
+		},
+		UpdatedAt: database.Now(),
+		Error: sql.NullString{
+			String: cancelJob.Error,
+			Valid:  cancelJob.Error != "",
+		},
+	})
+	if err != nil {
+		return nil, xerrors.Errorf("update provisioner job: %w", err)
+	}
+	return &proto.Empty{}, nil
+}
+
+// CompleteJob is triggered by a provision daemon to mark a provisioner job as completed.
+func (server *provisionerdServer) CompleteJob(ctx context.Context, completed *proto.CompletedJob) (*proto.Empty, error) {
+	jobID, err := uuid.Parse(completed.JobId)
+	if err != nil {
+		return nil, xerrors.Errorf("parse job id: %w", err)
+	}
+	job, err := server.Database.GetProvisionerJobByID(ctx, jobID)
+	if err != nil {
+		return nil, xerrors.Errorf("get job by id: %w", err)
+	}
+	// TODO: Check if the worker ID matches!
+	// If it doesn't, a provisioner daemon could be impersonating another job!
+
+	switch jobType := completed.Type.(type) {
+	case *proto.CompletedJob_ProjectImport_:
+		var input projectImportJob
+		err = json.Unmarshal(job.Input, &input)
+		if err != nil {
+			return nil, xerrors.Errorf("unmarshal job data: %w", err)
+		}
+
+		// Validate that all parameters send from the provisioner daemon
+		// follow the protocol.
+		projectParameters := make([]database.InsertProjectParameterParams, 0, len(jobType.ProjectImport.ParameterSchemas))
+		for _, protoParameter := range jobType.ProjectImport.ParameterSchemas {
+			validationTypeSystem, err := convertValidationTypeSystem(protoParameter.ValidationTypeSystem)
+			if err != nil {
+				return nil, xerrors.Errorf("convert validation type system for %q: %w", protoParameter.Name, err)
+			}
+
+			projectParameter := database.InsertProjectParameterParams{
+				ID:                   uuid.New(),
+				CreatedAt:            database.Now(),
+				ProjectHistoryID:     input.ProjectHistoryID,
+				Name:                 protoParameter.Name,
+				Description:          protoParameter.Description,
+				RedisplayValue:       protoParameter.RedisplayValue,
+				ValidationError:      protoParameter.ValidationError,
+				ValidationCondition:  protoParameter.ValidationCondition,
+				ValidationValueType:  protoParameter.ValidationValueType,
+				ValidationTypeSystem: validationTypeSystem,
+
+				AllowOverrideDestination: protoParameter.AllowOverrideDestination,
+				AllowOverrideSource:      protoParameter.AllowOverrideSource,
+			}
+
+			// It's possible a parameter doesn't define a default source!
+			if protoParameter.DefaultSource != nil {
+				parameterSourceScheme, err := convertParameterSourceScheme(protoParameter.DefaultSource.Scheme)
+				if err != nil {
+					return nil, xerrors.Errorf("convert parameter source scheme: %w", err)
+				}
+				projectParameter.DefaultSourceScheme = parameterSourceScheme
+				projectParameter.DefaultSourceValue = sql.NullString{
+					String: protoParameter.DefaultSource.Value,
+					Valid:  protoParameter.DefaultSource.Value != "",
+				}
+			}
+
+			// It's possible a parameter doesn't define a default destination!
+			if protoParameter.DefaultDestination != nil {
+				parameterDestinationScheme, err := convertParameterDestinationScheme(protoParameter.DefaultDestination.Scheme)
+				if err != nil {
+					return nil, xerrors.Errorf("convert parameter destination scheme: %w", err)
+				}
+				projectParameter.DefaultDestinationScheme = parameterDestinationScheme
+				projectParameter.DefaultDestinationValue = sql.NullString{
+					String: protoParameter.DefaultDestination.Value,
+					Valid:  protoParameter.DefaultDestination.Value != "",
+				}
+			}
+
+			projectParameters = append(projectParameters, projectParameter)
+		}
+
+		// This must occur in a transaction in case of failure.
+		err = server.Database.InTx(func(db database.Store) error {
+			err = db.UpdateProvisionerJobByID(ctx, database.UpdateProvisionerJobByIDParams{
+				ID:        jobID,
+				UpdatedAt: database.Now(),
+				CompletedAt: sql.NullTime{
+					Time:  database.Now(),
+					Valid: true,
+				},
+			})
+			if err != nil {
+				return xerrors.Errorf("update provisioner job: %w", err)
+			}
+			// This could be a bulk-insert operation to improve performance.
+			// See the "InsertWorkspaceHistoryLogs" query.
+			for _, projectParameter := range projectParameters {
+				_, err = db.InsertProjectParameter(ctx, projectParameter)
+				if err != nil {
+					return xerrors.Errorf("insert project parameter %q: %w", projectParameter.Name, err)
+				}
+			}
+			return nil
+		})
+		if err != nil {
+			return nil, xerrors.Errorf("complete job: %w", err)
+		}
+	case *proto.CompletedJob_WorkspaceProvision_:
+		var input workspaceProvisionJob
+		err = json.Unmarshal(job.Input, &input)
+		if err != nil {
+			return nil, xerrors.Errorf("unmarshal job data: %w", err)
+		}
+
+		workspaceHistory, err := server.Database.GetWorkspaceHistoryByID(ctx, input.WorkspaceHistoryID)
+		if err != nil {
+			return nil, xerrors.Errorf("get workspace history: %w", err)
+		}
+
+		err = server.Database.InTx(func(db database.Store) error {
+			err = db.UpdateProvisionerJobByID(ctx, database.UpdateProvisionerJobByIDParams{
+				ID:        jobID,
+				UpdatedAt: database.Now(),
+				CompletedAt: sql.NullTime{
+					Time:  database.Now(),
+					Valid: true,
+				},
+			})
+			if err != nil {
+				return xerrors.Errorf("update provisioner job: %w", err)
+			}
+			err = db.UpdateWorkspaceHistoryByID(ctx, database.UpdateWorkspaceHistoryByIDParams{
+				ID:               workspaceHistory.ID,
+				UpdatedAt:        database.Now(),
+				ProvisionerState: jobType.WorkspaceProvision.State,
+			})
+			if err != nil {
+				return xerrors.Errorf("update workspace history: %w", err)
+			}
+			// This could be a bulk insert to improve performance.
+			for _, protoResource := range jobType.WorkspaceProvision.Resources {
+				_, err = db.InsertWorkspaceResource(ctx, database.InsertWorkspaceResourceParams{
+					ID:                 uuid.New(),
+					CreatedAt:          database.Now(),
+					WorkspaceHistoryID: input.WorkspaceHistoryID,
+					Type:               protoResource.Type,
+					Name:               protoResource.Name,
+					// TODO: Generate this at the variable validation phase.
+					// Set the value in `default_source`, and disallow overwrite.
+					WorkspaceAgentToken: uuid.NewString(),
+				})
+				if err != nil {
+					return xerrors.Errorf("insert workspace resource %q: %w", protoResource.Name, err)
+				}
+			}
+			return nil
+		})
+		if err != nil {
+			return nil, xerrors.Errorf("complete job: %w", err)
+		}
+	default:
+		return nil, xerrors.Errorf("unknown job type %q; ensure coderd and provisionerd versions match",
+			reflect.TypeOf(completed.Type).String())
+	}
+
+	return &proto.Empty{}, nil
+}
+
+func convertValidationTypeSystem(typeSystem sdkproto.ParameterSchema_TypeSystem) (database.ParameterTypeSystem, error) {
+	switch typeSystem {
+	case sdkproto.ParameterSchema_HCL:
+		return database.ParameterTypeSystemHCL, nil
+	default:
+		return database.ParameterTypeSystem(""), xerrors.Errorf("unknown type system: %d", typeSystem)
+	}
+}
+
+func convertParameterSourceScheme(sourceScheme sdkproto.ParameterSource_Scheme) (database.ParameterSourceScheme, error) {
+	switch sourceScheme {
+	case sdkproto.ParameterSource_DATA:
+		return database.ParameterSourceSchemeData, nil
+	default:
+		return database.ParameterSourceScheme(""), xerrors.Errorf("unknown parameter source scheme: %d", sourceScheme)
+	}
+}
+
+func convertParameterDestinationScheme(destinationScheme sdkproto.ParameterDestination_Scheme) (database.ParameterDestinationScheme, error) {
+	switch destinationScheme {
+	case sdkproto.ParameterDestination_ENVIRONMENT_VARIABLE:
+		return database.ParameterDestinationSchemeEnvironmentVariable, nil
+	case sdkproto.ParameterDestination_PROVISIONER_VARIABLE:
+		return database.ParameterDestinationSchemeProvisionerVariable, nil
+	default:
+		return database.ParameterDestinationScheme(""), xerrors.Errorf("unknown parameter destination scheme: %d", destinationScheme)
+	}
+}
+
+func convertLogLevel(logLevel sdkproto.LogLevel) (database.LogLevel, error) {
+	switch logLevel {
+	case sdkproto.LogLevel_TRACE:
+		return database.LogLevelTrace, nil
+	case sdkproto.LogLevel_DEBUG:
+		return database.LogLevelDebug, nil
+	case sdkproto.LogLevel_INFO:
+		return database.LogLevelInfo, nil
+	case sdkproto.LogLevel_WARN:
+		return database.LogLevelWarn, nil
+	case sdkproto.LogLevel_ERROR:
+		return database.LogLevelError, nil
+	default:
+		return database.LogLevel(""), xerrors.Errorf("unknown log level: %d", logLevel)
+	}
+}
+
+func convertLogSource(logSource proto.LogSource) (database.LogSource, error) {
+	switch logSource {
+	case proto.LogSource_PROVISIONER_DAEMON:
+		return database.LogSourceProvisionerDaemon, nil
+	case proto.LogSource_PROVISIONER:
+		return database.LogSourceProvisioner, nil
+	default:
+		return database.LogSource(""), xerrors.Errorf("unknown log source: %d", logSource)
+	}
+}