Function reapJob

coderd/jobreaper/detector.go:237–402

(ctx context.Context, log slog.Logger, db database.Store, pub pubsub.Pubsub, jobToReap *jobToReap)

}

func reapJob(ctx context.Context, log slog.Logger, db database.Store, pub pubsub.Pubsub, jobToReap *jobToReap) error {
	var lowestLogID int64

	err := db.InTx(func(db database.Store) error {
		// Refetch the job while we hold the lock.
		job, err := db.GetProvisionerJobByIDForUpdate(ctx, jobToReap.ID)
		if err != nil {
			if xerrors.Is(err, sql.ErrNoRows) {
				return acquireLockError{}
			}
			return xerrors.Errorf("get provisioner job: %w", err)
		}

		if job.CompletedAt.Valid {
			return jobIneligibleError{
				Err: xerrors.Errorf("job is completed (status %s)", job.JobStatus),
			}
		}
		if job.UpdatedAt.After(time.Now().Add(-jobToReap.Threshold)) {
			return jobIneligibleError{
				Err: xerrors.New("job has been updated recently"),
			}
		}

		log.Warn(
			ctx, "forcefully terminating provisioner job",
			slog.F("type", jobToReap.Type),
			slog.F("threshold", jobToReap.Threshold),
		)

		// First, get the latest logs from the build so we can make sure
		// our messages are in the latest stage.
		logs, err := db.GetProvisionerLogsAfterID(ctx, database.GetProvisionerLogsAfterIDParams{
			JobID:        job.ID,
			CreatedAfter: 0,
		})
		if err != nil {
			return xerrors.Errorf("get logs for %s job: %w", jobToReap.Type, err)
		}
		logStage := ""
		if len(logs) != 0 {
			logStage = logs[len(logs)-1].Stage
		}
		if logStage == "" {
			logStage = "Unknown"
		}

		// Insert the messages into the build log.
		insertParams := database.InsertProvisionerJobLogsParams{
			JobID:     job.ID,
			CreatedAt: nil,
			Source:    nil,
			Level:     nil,
			Stage:     nil,
			Output:    nil,
		}
		now := dbtime.Now()
		for i, msg := range JobLogMessages(jobToReap.Type, jobToReap.Threshold) {