Repository: osbuild/osbuild-composer
Pull request #595 (2020-05-10 17:36:22)

jobqueue follow ups, including cancelation
This series includes a few follow-up commits for the jobqueue PR. Some is cleaning up, some is refactoring to make it possible to support cancelation (the last commit).

larskarlitski: Rebased to include the schutzbot fixes.

Patch:
diff --git a/cmd/osbuild-composer/main.go b/cmd/osbuild-composer/main.go
index 8a7cbeb32..338049fc5 100644
--- a/cmd/osbuild-composer/main.go
+++ b/cmd/osbuild-composer/main.go
@@ -125,7 +125,7 @@ func main() {
 		log.Fatalf("cannot create queue directory: %v", err)
 	}
 
-	jobs, err := fsjobqueue.New(queueDir)
+	jobs, err := fsjobqueue.New(queueDir, []string{"osbuild"})
 	if err != nil {
 		log.Fatalf("cannot create jobqueue: %v", err)
 	}
diff --git a/cmd/osbuild-worker/main.go b/cmd/osbuild-worker/main.go
index ce1c856cf..d80cf65f7 100644
--- a/cmd/osbuild-worker/main.go
+++ b/cmd/osbuild-worker/main.go
@@ -64,14 +64,14 @@ func (e *TargetsError) Error() string {
 }
 
 func RunJob(job *worker.Job, uploadFunc func(uuid.UUID, int, io.Reader) error) (*common.ComposeResult, error) {
-	tmpStore, err := ioutil.TempDir("/var/tmp", "osbuild-store")
+	tmpOutput, err := ioutil.TempDir("/var/tmp", "osbuild-output-*")
 	if err != nil {
-		return nil, fmt.Errorf("error setting up osbuild store: %v", err)
+		return nil, fmt.Errorf("error setting up osbuild output directory: %v", err)
 	}
 	// FIXME: how to handle errors in defer?
-	defer os.RemoveAll(tmpStore)
+	defer os.RemoveAll(tmpOutput)
 
-	result, err := RunOSBuild(job.Manifest, tmpStore, os.Stderr)
+	result, err := RunOSBuild(job.Manifest, tmpOutput, os.Stderr)
 	if err != nil {
 		return nil, err
 	}
@@ -81,7 +81,7 @@ func RunJob(job *worker.Job, uploadFunc func(uuid.UUID, int, io.Reader) error) (
 	for _, t := range job.Targets {
 		switch options := t.Options.(type) {
 		case *target.LocalTargetOptions:
-			f, err := os.Open(path.Join(tmpStore, "refs", result.OutputID, options.Filename))
+			f, err := os.Open(path.Join(tmpOutput, options.Filename))
 			if err != nil {
 				r = append(r, err)
 				continue
@@ -105,7 +105,7 @@ func RunJob(job *worker.Job, uploadFunc func(uuid.UUID, int, io.Reader) error) (
 				options.Key = job.Id.String()
 			}
 
-			_, err = a.Upload(path.Join(tmpStore, "refs", result.OutputID, options.Filename), options.Bucket, options.Key)
+			_, err = a.Upload(path.Join(tmpOutput, options.Filename), options.Bucket, options.Key)
 			if err != nil {
 				r = append(r, err)
 				continue
@@ -132,7 +132,7 @@ func RunJob(job *worker.Job, uploadFunc func(uuid.UUID, int, io.Reader) error) (
 			err := azure.UploadImage(
 				credentials,
 				metadata,
-				path.Join(tmpStore, "refs", result.OutputID, options.Filename),
+				path.Join(tmpOutput, options.Filename),
 				azureMaxUploadGoroutines,
 			)
 
diff --git a/cmd/osbuild-worker/osbuild.go b/cmd/osbuild-worker/osbuild.go
index b8a32da21..2da5dd167 100644
--- a/cmd/osbuild-worker/osbuild.go
+++ b/cmd/osbuild-worker/osbuild.go
@@ -19,10 +19,10 @@ func (e *OSBuildError) Error() string {
 	return e.Message
 }
 
-func RunOSBuild(manifest *osbuild.Manifest, store string, errorWriter io.Writer) (*common.ComposeResult, error) {
+func RunOSBuild(manifest *osbuild.Manifest, outputDirectory string, errorWriter io.Writer) (*common.ComposeResult, error) {
 	cmd := exec.Command(
 		"osbuild",
-		"--store", store,
+		"--output-directory", outputDirectory,
 		"--json", "-",
 	)
 	cmd.Stderr = errorWriter
diff --git a/internal/compose/compose.go b/internal/compose/compose.go
index ac56a09fd..a6811534e 100644
--- a/internal/compose/compose.go
+++ b/internal/compose/compose.go
@@ -100,29 +100,3 @@ func (c *Compose) DeepCopy() Compose {
 		ImageBuilds: newImageBuilds,
 	}
 }
-
-// UpdateState changes a state of a single image build inside the Compose
-func (c *Compose) UpdateState(imageBuildId int, newState common.ImageBuildState) error {
-	switch newState {
-	case common.IBWaiting:
-		return &StateTransitionError{"image build cannot be moved into waiting state"}
-	case common.IBRunning:
-		if c.ImageBuilds[imageBuildId].QueueStatus == common.IBWaiting || c.ImageBuilds[imageBuildId].QueueStatus == common.IBRunning {
-			c.ImageBuilds[imageBuildId].QueueStatus = newState
-		} else {
-			return &StateTransitionError{"only waiting image build can be transitioned into running state"}
-		}
-	case common.IBFinished, common.IBFailed:
-		if c.ImageBuilds[imageBuildId].QueueStatus == common.IBRunning {
-			c.ImageBuilds[imageBuildId].QueueStatus = newState
-			for _, t := range c.ImageBuilds[imageBuildId].Targets {
-				t.Status = newState
-			}
-		} else {
-			return &StateTransitionError{"only running image build can be transitioned into finished or failed state"}
-		}
-	default:
-		return &StateTransitionError{"invalid state"}
-	}
-	return nil
-}
diff --git a/internal/jobqueue/fsjobqueue/fsjobqueue.go b/internal/jobqueue/fsjobqueue/fsjobqueue.go
index 76443b8cd..7a407d95d 100644
--- a/internal/jobqueue/fsjobqueue/fsjobqueue.go
+++ b/internal/jobqueue/fsjobqueue/fsjobqueue.go
@@ -3,7 +3,7 @@
 //
 // Jobs are stored in the file system, using the `jsondb` package. However,
 // this package does not use the file system as a database, but keeps some
-// state in memory. This means that access to a given directory myst be
+// state in memory. This means that access to a given directory must be
 // exclusive to only one `fsJobQueue` object at a time. A single `fsJobQueue`
 // can be safely accessed from multiple goroutines, though.
 //
@@ -14,6 +14,7 @@ package fsjobqueue
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"reflect"
 	"sort"
@@ -26,19 +27,20 @@ import (
 )
 
 type fsJobQueue struct {
+	// Protects all fields of this struct. In particular, it ensures
+	// transactions on `db` are atomic. All public functions except
+	// JobStatus hold it while they're running. Dequeue() releases it
+	// briefly while waiting on pending channels.
+	mu sync.Mutex
+
 	db *jsondb.JSONDatabase
 
-	// Maps job types to channels of job ids for that type. Only access
-	// through pendingChannel() to ensure concurrent access is restricted
-	// by the mutex.
-	pending      map[string]chan uuid.UUID
-	pendingMutex sync.Mutex
+	// Maps job types to channels of job ids for that type.
+	pending map[string]chan uuid.UUID
 
 	// Maps job ids to the jobs that depend on it, if any of those
-	// dependants have not yet finished. Only acccess while holding the
-	// associated mutex.
-	dependants      map[uuid.UUID][]uuid.UUID
-	dependantsMutex sync.Mutex
+	// dependants have not yet finished.
+	dependants map[uuid.UUID][]uuid.UUID
 }
 
 // On-disk job struct. Contains all necessary (but non-redundant) information
@@ -51,22 +53,27 @@ type job struct {
 	Dependencies []uuid.UUID     `json:"dependencies"`
 	Result       json.RawMessage `json:"result,omitempty"`
 
-	Status     jobqueue.JobStatus `json:"status"`
-	QueuedAt   time.Time          `json:"queued-at,omitempty"`
-	StartedAt  time.Time          `json:"started-at,omitempty"`
-	FinishedAt time.Time          `json:"finished-at,omitempty"`
+	QueuedAt   time.Time `json:"queued_at,omitempty"`
+	StartedAt  time.Time `json:"started_at,omitempty"`
+	FinishedAt time.Time `json:"finished_at,omitempty"`
+
+	Canceled bool `json:"canceled,omitempty"`
 }
 
 // Create a new fsJobQueue object for `dir`. This object must have exclusive
 // access to `dir`. If `dir` contains jobs created from previous runs, they are
 // loaded and rescheduled to run if necessary.
-func New(dir string) (*fsJobQueue, error) {
+func New(dir string, acceptedJobTypes []string) (*fsJobQueue, error) {
 	q := &fsJobQueue{
 		db:         jsondb.New(dir, 0600),
 		pending:    make(map[string]chan uuid.UUID),
 		dependants: make(map[uuid.UUID][]uuid.UUID),
 	}
 
+	for _, jt := range acceptedJobTypes {
+		q.pending[jt] = make(chan uuid.UUID, 100)
+	}
+
 	// Look for jobs that are still pending and build the dependant map.
 	ids, err := q.db.List()
 	if err != nil {
@@ -81,36 +88,27 @@ func New(dir string) (*fsJobQueue, error) {
 		if err != nil {
 			return nil, err
 		}
-		// We only enqueue jobs that were previously pending.
-		if j.Status != jobqueue.JobPending {
-			continue
-		}
-		// Initialize dependants for this job.
-		q.dependantsMutex.Lock()
-		defer q.dependantsMutex.Unlock()
-		for _, dep := range j.Dependencies {
-			q.dependants[dep] = append(q.dependants[dep], j.Id)
-		}
-		// Enqueue a job if all its dependencies (if there are any)
-		// have finished, but the job itself hasn't run yet.
-		n, err := q.countFinishedJobs(j.Dependencies)
+		err = q.maybeEnqueue(j, true)
 		if err != nil {
 			return nil, err
 		}
-		if n == len(j.Dependencies) {
-			q.pendingChannel(j.Type) <- j.Id
-		}
 	}
 
 	return q, nil
 }
 
 func (q *fsJobQueue) Enqueue(jobType string, args interface{}, dependencies []uuid.UUID) (uuid.UUID, error) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	if _, exists := q.pending[jobType]; !exists {
+		return uuid.Nil, fmt.Errorf("this queue does not accept job type '%s'", jobType)
+	}
+
 	var j = job{
 		Id:           uuid.New(),
 		Type:         jobType,
 		Dependencies: uniqueUUIDList(dependencies),
-		Status:       jobqueue.JobPending,
 		QueuedAt:     time.Now(),
 	}
 
@@ -120,10 +118,16 @@ func (q *fsJobQueue) Enqueue(jobType string, args interface{}, dependencies []uu
 		return uuid.Nil, fmt.Errorf("error marshaling job arguments: %v", err)
 	}
 
-	// Verify dependencies and check how many of them are already finished.
-	finished, err := q.countFinishedJobs(j.Dependencies)
-	if err != nil {
-		return uuid.Nil, err
+	// Verify dependendencies early, so that the job doesn't get written
+	// when one of them doesn't exist.
+	for _, d := range j.Dependencies {
+		exists, err := q.db.Read(d.String(), nil)
+		if err != nil {
+			return uuid.Nil, err
+		}
+		if !exists {
+			return uuid.Nil, jobqueue.ErrNotExist
+		}
 	}
 
 	// Write the job before updating in-memory state, so that the latter
@@ -133,85 +137,87 @@ func (q *fsJobQueue) Enqueue(jobType string, args interface{}, dependencies []uu
 		return uuid.Nil, fmt.Errorf("cannot write job: %v:", err)
 	}
 
-	// If all dependencies have finished, or there are none, queue the job.
-	// Otherwise, update dependants so that this check is done again when
-	// FinishJob() is called for a dependency.
-	if finished == len(j.Dependencies) {
-		q.pendingChannel(j.Type) <- j.Id
-	} else {
-		q.dependantsMutex.Lock()
-		defer q.dependantsMutex.Unlock()
-		for _, id := range j.Dependencies {
-			q.dependants[id] = append(q.dependants[id], j.Id)
-		}
+	err = q.maybeEnqueue(&j, true)
+	if err != nil {
+		return uuid.Nil, err
 	}
 
 	return j.Id, nil
 }
 
 func (q *fsJobQueue) Dequeue(ctx context.Context, jobTypes []string, args interface{}) (uuid.UUID, error) {
-	// Return early if the conext is already canceled.
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	// Return early if the context is already canceled.
 	if err := ctx.Err(); err != nil {
 		return uuid.Nil, err
 	}
 
-	// Use reflect.Select(), because the `select` statement cannot operate
-	// on an unknown amount of channels.
-	cases := []reflect.SelectCase{
-		{
-			Dir:  reflect.SelectRecv,
-			Chan: reflect.ValueOf(ctx.Done()),
-		},
-	}
+	// Filter q.pending by the `jobTypes`. Ignore those job types that this
+	// queue doesn't accept.
+	chans := []chan uuid.UUID{}
 	for _, jt := range jobTypes {
-		cases = append(cases, reflect.SelectCase{
-			Dir:  reflect.SelectRecv,
-			Chan: reflect.ValueOf(q.pendingChannel(jt)),
-		})
+		if c, exists := q.pending[jt]; exists {
+			chans = append(chans, c)
+		}
 	}
 
-	chosen, value, recvOK := reflect.Select(cases)
-	if chosen == 0 && !recvOK {
-		return uuid.Nil, ctx.Err()
-	}
-	// pending channels are never closed
-	if !recvOK {
-		panic("pending channel was closed unexpectedly")
-	}
-	id := value.Interface().(uuid.UUID)
+	// Loop until finding a non-canceled job.
+	var j *job
+	for {
+		// Unlock the mutex while polling channels, so that multiple goroutines
+		// can wait at the same time.
+		q.mu.Unlock()
+		id, err := selectUUIDChannel(ctx, chans)
+		q.mu.Lock()
 
-	j, err := q.readJob(id)
-	if err != nil {
-		return uuid.Nil, err
+		if err != nil {
+			return uuid.Nil, err
+		}
+
+		j, err = q.readJob(id)
+		if err != nil {
+			return uuid.Nil, err
+		}
+
+		if !j.Canceled {
+			break
+		}
 	}
 
-	err = json.Unmarshal(j.Args, args)
+	err := json.Unmarshal(j.Args, args)
 	if err != nil {
 		return uuid.Nil, fmt.Errorf("error unmarshaling arguments for job '%s': %v", j.Id, err)
 	}
 
-	j.Status = jobqueue.JobRunning
 	j.StartedAt = time.Now()
 
-	err = q.db.Write(id.String(), j)
+	err = q.db.Write(j.Id.String(), j)
 	if err != nil {
-		return uuid.Nil, fmt.Errorf("error writing job %s: %v", id, err)
+		return uuid.Nil, fmt.Errorf("error writing job %s: %v", j.Id, err)
 	}
 
 	return j.Id, nil
 }
 
 func (q *fsJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
 	j, err := q.readJob(id)
 	if err != nil {
 		return err
 	}
 
-	if j.Status != jobqueue.JobRunning {
+	if j.Canceled {
+		return jobqueue.ErrCanceled
+	}
+
+	if j.StartedAt.IsZero() || !j.FinishedAt.IsZero() {
 		return jobqueue.ErrNotRunning
 	}
 
-	j.Status = jobqueue.JobFinished
 	j.FinishedAt = time.Now()
 
 	j.Result, err = json.Marshal(result)
@@ -225,35 +231,51 @@ func (q *fsJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
 		return fmt.Errorf("error writing job %s: %v", id, err)
 	}
 
-	q.dependantsMutex.Lock()
-	defer q.dependantsMutex.Unlock()
 	for _, depid := range q.dependants[id] {
 		dep, err := q.readJob(depid)
 		if err != nil {
 			return err
 		}
-		n, err := q.countFinishedJobs(dep.Dependencies)
+		err = q.maybeEnqueue(dep, false)
 		if err != nil {
 			return err
 		}
-		if n == len(dep.Dependencies) {
-			q.pendingChannel(dep.Type) <- dep.Id
-		}
 	}
 	delete(q.dependants, id)
 
 	return nil
 }
 
-func (q *fsJobQueue) JobStatus(id uuid.UUID, result interface{}) (status jobqueue.JobStatus, queued, started, finished time.Time, err error) {
-	var j *job
+func (q *fsJobQueue) CancelJob(id uuid.UUID) error {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	j, err := q.readJob(id)
+	if err != nil {
+		return err
+	}
+
+	if !j.FinishedAt.IsZero() {
+		return nil
+	}
+
+	j.Canceled = true
+
+	err = q.db.Write(id.String(), j)
+	if err != nil {
+		return fmt.Errorf("error writing job %s: %v", id, err)
+	}
 
-	j, err = q.readJob(id)
+	return nil
+}
+
+func (q *fsJobQueue) JobStatus(id uuid.UUID, result interface{}) (queued, started, finished time.Time, canceled bool, err error) {
+	j, err := q.readJob(id)
 	if err != nil {
 		return
 	}
 
-	if j.Status == jobqueue.JobFinished {
+	if !j.FinishedAt.IsZero() && !j.Canceled {
 		err = json.Unmarshal(j.Result, result)
 		if err != nil {
 			err = fmt.Errorf("error unmarshaling result for job '%s': %v", id, err)
@@ -261,30 +283,14 @@ func (q *fsJobQueue) JobStatus(id uuid.UUID, result interface{}) (status jobqueu
 		}
 	}
 
-	status = j.Status
 	queued = j.QueuedAt
 	started = j.StartedAt
 	finished = j.FinishedAt
+	canceled = j.Canceled
 
 	return
 }
 
-// Returns the number of finished jobs in `ids`.
-func (q *fsJobQueue) countFinishedJobs(ids []uuid.UUID) (int, error) {
-	n := 0
-	for _, id := range ids {
-		j, err := q.readJob(id)
-		if err != nil {
-			return 0, err
-		}
-		if j.Status == jobqueue.JobFinished {
-			n += 1
-		}
-	}
-
-	return n, nil
-}
-
 // Reads job with `id`. This is a thin wrapper around `q.db.Read`, which
 // returns the job directly, or and error if a job with `id` does not exist.
 func (q *fsJobQueue) readJob(id uuid.UUID) (*job, error) {
@@ -300,19 +306,39 @@ func (q *fsJobQueue) readJob(id uuid.UUID) (*job, error) {
 	return &j, nil
 }
 
-// Safe access to the pending channel for `jobType`. Channels are created on
-// demand.
-func (q *fsJobQueue) pendingChannel(jobType string) chan uuid.UUID {
-	q.pendingMutex.Lock()
-	defer q.pendingMutex.Unlock()
+// Enqueue `job` if it is pending and all its dependencies have finished.
+// Update `q.dependants` if the job was not queued and updateDependants is true
+// (i.e., when this is a new job).
+func (q *fsJobQueue) maybeEnqueue(j *job, updateDependants bool) error {
+	if !j.StartedAt.IsZero() {
+		return nil
+	}
 
-	c, exists := q.pending[jobType]
-	if !exists {
-		c = make(chan uuid.UUID, 100)
-		q.pending[jobType] = c
+	depsFinished := true
+	for _, id := range j.Dependencies {
+		j, err := q.readJob(id)
+		if err != nil {
+			return err
+		}
+		if j.FinishedAt.IsZero() {
+			depsFinished = false
+			break
+		}
+	}
+
+	if depsFinished {
+		c, exists := q.pending[j.Type]
+		if !exists {
+			return fmt.Errorf("this queue doesn't accept job type '%s'", j.Type)
+		}
+		c <- j.Id
+	} else if updateDependants {
+		for _, id := range j.Dependencies {
+			q.dependants[id] = append(q.dependants[id], j.Id)
+		}
 	}
 
-	return c
+	return nil
 }
 
 // Sorts and removes duplicates from `ids`.
@@ -338,3 +364,34 @@ func uniqueUUIDList(ids []uuid.UUID) []uuid.UUID {
 
 	return l
 }
+
+// Select on a list of `chan uuid.UUID`s. Returns an error if one of the
+// channels is closed.
+//
+// Uses reflect.Select(), because the `select` statement cannot operate on an
+// unknown amount of channels.
+func selectUUIDChannel(ctx context.Context, chans []chan uuid.UUID) (uuid.UUID, error) {
+	cases := []reflect.SelectCase{
+		{
+			Dir:  reflect.SelectRecv,
+			Chan: reflect.ValueOf(ctx.Done()),
+		},
+	}
+	for _, c := range chans {
+		cases = append(cases, reflect.SelectCase{
+			Dir:  reflect.SelectRecv,
+			Chan: reflect.ValueOf(c),
+		})
+	}
+
+	chosen, value, recvOK := reflect.Select(cases)
+	if !recvOK {
+		if chosen == 0 {
+			return uuid.Nil, ctx.Err()
+		} else {
+			return uuid.Nil, errors.New("channel was closed unexpectedly")
+		}
+	}
+
+	return value.Interface().(uuid.UUID), nil
+}
diff --git a/internal/jobqueue/jobqueue.go b/internal/jobqueue/jobqueue.go
index a33a32c1a..6dbd96b0e 100644
--- a/internal/jobqueue/jobqueue.go
+++ b/internal/jobqueue/jobqueue.go
@@ -14,7 +14,6 @@ package jobqueue
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"time"
 
@@ -49,57 +48,20 @@ type JobQueue interface {
 	// job type and must be serializable to JSON.
 	FinishJob(id uuid.UUID, result interface{}) error
 
-	// Returns the current status of the job. If the job has already
-	// finished, its result will be returned in `result`. Also returns the
-	// time the job was
-	//    queued   - always valid
-	//    started  - valid when the job is running or has finished
-	//    finished - valid when the job has finished
-	JobStatus(id uuid.UUID, result interface{}) (status JobStatus, queued, started, finished time.Time, err error)
-}
-
-type JobStatus int
+	// Cancel a job. Does nothing if the job has already finished.
+	CancelJob(id uuid.UUID) error
 
-const (
-	JobPending JobStatus = iota
-	JobRunning
-	JobFinished
-)
-
-func (s JobStatus) String() string {
-	switch s {
-	case JobPending:
-		return "pending"
-	case JobRunning:
-		return "running"
-	case JobFinished:
-		return "finished"
-	default:
-		return "<invalid>"
-	}
-}
-
-func (s JobStatus) MarshalJSON() ([]byte, error) {
-	return json.Marshal(s.String())
-}
-
-func (s *JobStatus) UnmarshalJSON(data []byte) error {
-	var str string
-	if err := json.Unmarshal(data, &str); err != nil {
-		return err
-	}
-	switch str {
-	case "pending":
-		*s = JobPending
-	case "running":
-		*s = JobRunning
-	case "finished":
-		*s = JobFinished
-	}
-	return nil
+	// Returns the current status of the job, in the form of three times:
+	// queued, started, and finished. `started` and `finished` might be the
+	// zero time (check with t.IsZero()), when the job is not running or
+	// finished, respectively.
+	//
+	// If the job is finished, its result will be returned in `result`.
+	JobStatus(id uuid.UUID, result interface{}) (queued, started, finished time.Time, canceled bool, err error)
 }
 
 var (
 	ErrNotExist   = errors.New("job does not exist")
 	ErrNotRunning = errors.New("job is not running")
+	ErrCanceled   = errors.New("job ws canceled")
 )
diff --git a/internal/jsondb/db.go b/internal/jsondb/db.go
index 6551f4ed1..10cef85c1 100644
--- a/internal/jsondb/db.go
+++ b/internal/jsondb/db.go
@@ -36,7 +36,8 @@ func New(dir string, perm os.FileMode) *JSONDatabase {
 	return &JSONDatabase{dir, perm}
 }
 
-// Reads the value at `name`. `document` must deserializable from JSON. Returns
+// Reads the value at `name`. `document` must be a type that is deserializable
+// from the JSON document `name`, or nil to not deserialize at all. Returns
 // false if a document with `name` does not exist.
 func (db *JSONDatabase) Read(name string, document interface{}) (bool, error) {
 	f, err := os.Open(path.Join(db.dir, name+".json"))
@@ -48,9 +49,11 @@ func (db *JSONDatabase) Read(name string, document interface{}) (bool, error) {
 	}
 	defer f.Close()
 
-	err = json.NewDecoder(f).Decode(&document)
-	if err != nil {
-		return false, fmt.Errorf("error reading db file %s: %v", name, err)
+	if document != nil {
+		err = json.NewDecoder(f).Decode(&document)
+		if err != nil {
+			return false, fmt.Errorf("error reading db file %s: %v", name, err)
+		}
 	}
 
 	return true, nil
diff --git a/internal/store/store.go b/internal/store/store.go
index a10e1aa79..03f07a538 100644
--- a/internal/store/store.go
+++ b/internal/store/store.go
@@ -76,22 +76,6 @@ func (e *NotFoundError) Error() string {
 	return e.message
 }
 
-type NotPendingError struct {
-	message string
-}
-
-func (e *NotPendingError) Error() string {
-	return e.message
-}
-
-type InvalidRequestError struct {
-	message string
-}
-
-func (e *InvalidRequestError) Error() string {
-	return e.message
-}
-
 type NoLocalTargetError struct {
 	message string
 }
@@ -554,6 +538,22 @@ func (s *Store) PushTestCompose(composeID uuid.UUID, manifest *osbuild.Manifest,
 		if err != nil {
 			return fmt.Errorf("cannot create output directory for job %v: %#v", composeID, err)
 		}
+
+		f, err := os.Create(s.getImageBuildDirectory(composeID, 0) + "/result.json")
+		if err != nil {
+			return fmt.Errorf("cannot open result.json for job %v: %#v", composeID, err)
+		}
+		err = json.NewEncoder(f).Encode(common.ComposeResult{Success: testSuccess})
+		if err != nil {
+			return fmt.Errorf("cannot write to result.json for job %v: %#v", composeID, err)
+		}
+	}
+
+	var status common.ImageBuildState
+	if testSuccess {
+		status = common.IBFinished
+	} else {
+		status = common.IBFailed
 	}
 
 	// FIXME: handle or comment this possible error
@@ -562,12 +562,13 @@ func (s *Store) PushTestCompose(composeID uuid.UUID, manifest *osbuild.Manifest,
 			Blueprint: bp,
 			ImageBuilds: []compose.ImageBuild{
 				{
-					QueueStatus: common.IBRunning,
+					QueueStatus: status,
 					Manifest:    manifest,
 					ImageType:   imageTypeCommon,
 					Targets:     targets,
 					JobCreated:  time.Now(),
 					JobStarted:  time.Now(),
+					JobFinished: time.Now(),
 					Size:        size,
 				},
 			},
@@ -575,22 +576,6 @@ func (s *Store) PushTestCompose(composeID uuid.UUID, manifest *osbuild.Manifest,
 		return nil
 	})
 
-	var status common.ImageBuildState
-	var result common.ComposeResult
-	if testSuccess {
-		status = common.IBFinished
-		result = common.ComposeResult{Success: true}
-	} else {
-		status = common.IBFailed
-		result = common.ComposeResult{}
-	}
-
-	// Instead of starting the job, immediately set a final status
-	err := s.UpdateImageBuildInCompose(composeID, 0, status, &result)
-	if err != nil {
-		return err
-	}
-
 	return nil
 }
 
@@ -616,49 +601,6 @@ func (s *Store) DeleteCompose(id uuid.UUID) error {
 	})
 }
 
-// UpdateImageBuildInCompose sets the status and optionally also the final image.
-func (s *Store) UpdateImageBuildInCompose(composeID uuid.UUID, imageBuildID int, status common.ImageBuildState, result *common.ComposeResult) error {
-	return s.change(func() error {
-		// Check that the compose exists
-		currentCompose, exists := s.Composes[composeID]
-		if !exists {
-			return &NotFoundError{"compose does not exist"}
-		}
-		// Check that the image build was waiting
-		if currentCompose.ImageBuilds[imageBuildID].QueueStatus == common.IBWaiting {
-			return &NotPendingError{"compose has not been popped"}
-		}
-
-		// write result into file
-		if s.stateDir != nil && result != nil {
-			f, err := os.Create(s.getImageBuildDirectory(composeID, imageBuildID) + "/result.json")
-
-			if err != nil {
-				return fmt.Errorf("cannot open result.json for job %v: %#v", composeID, err)
-			}
-
-			// FIXME: handle error
-			_ = json.NewEncoder(f).Encode(result)
-		}
-
-		// Update the image build state including all target states
-		err := currentCompose.UpdateState(imageBuildID, status)
-		if err != nil {
-			// TODO: log error
-			return &InvalidRequestError{"invalid state transition: " + err.Error()}
-		}
-
-		// In case the image build is done, store the time and possibly also the image
-		if status == common.IBFinished || status == common.IBFailed {
-			currentCompose.ImageBuilds[imageBuildID].JobFinished = time.Now()
-		}
-
-		s.Composes[composeID] = currentCompose
-
-		return nil
-	})
-}
-
 func (s *Store) AddImageToImageUpload(composeID uuid.UUID, imageBuildID int, reader io.Reader) error {
 	currentCompose, exists := s.Composes[composeID]
 	if !exists {
diff --git a/internal/worker/server.go b/internal/worker/server.go
index b13345c8e..768120c14 100644
--- a/internal/worker/server.go
+++ b/internal/worker/server.go
@@ -79,26 +79,54 @@ func (s *Server) Enqueue(manifest *osbuild.Manifest, targets []*target.Target) (
 }
 
 func (s *Server) JobStatus(id uuid.UUID) (state common.ComposeState, queued, started, finished time.Time, err error) {
+	var canceled bool
 	var result OSBuildJobResult
-	var status jobqueue.JobStatus
 
-	status, queued, started, finished, err = s.jobs.JobStatus(id, &result)
+	queued, started, finished, canceled, err = s.jobs.JobStatus(id, &result)
 	if err != nil {
 		return
 	}
 
-	state = composeStateFromJobStatus(status, result.OSBuildOutput)
+	if canceled {
+		state = common.CFailed
+	} else if !finished.IsZero() {
+		if result.OSBuildOutput.Success {
+			state = common.CFinished
+		} else {
+			state = common.CFailed
+		}
+	} else if !started.IsZero() {
+		state = common.CRunning
+	} else {
+		state = common.CWaiting
+	}
+
 	return
 }
 
 func (s *Server) JobResult(id uuid.UUID) (common.ComposeState, *common.ComposeResult, error) {
+	var canceled bool
 	var result OSBuildJobResult
-	status, _, _, _, err := s.jobs.JobStatus(id, &result)
+
+	_, started, finished, canceled, err := s.jobs.JobStatus(id, &result)
 	if err != nil {
 		return common.CWaiting, nil, err
 	}
 
-	return composeStateFromJobStatus(status, result.OSBuildOutput), result.OSBuildOutput, nil
+	state := common.CWaiting
+	if canceled {
+		state = common.CFailed
+	} else if !finished.IsZero() {
+		if result.OSBuildOutput.Success {
+			state = common.CFinished
+		} else {
+			state = common.CFailed
+		}
+	} else if !started.IsZero() {
+		state = common.CRunning
+	}
+
+	return state, result.OSBuildOutput, nil
 }
 
 // jsonErrorf() is similar to http.Error(), but returns the message in a json
@@ -216,19 +244,3 @@ func (s *Server) addJobImageHandler(writer http.ResponseWriter, request *http.Re
 		jsonErrorf(writer, http.StatusInternalServerError, "%v", err)
 	}
 }
-
-func composeStateFromJobStatus(status jobqueue.JobStatus, output *common.ComposeResult) common.ComposeState {
-	switch status {
-	case jobqueue.JobPending:
-		return common.CWaiting
-	case jobqueue.JobRunning:
-		return common.CRunning
-	case jobqueue.JobFinished:
-		if output.Success {
-			return common.CFinished
-		} else {
-			return common.CFailed
-		}
-	}
-	return common.CWaiting
-}


Test patch:
diff --git a/cmd/osbuild-image-tests/constants/constants-travis.go b/cmd/osbuild-image-tests/constants/constants-travis.go
index 48872e450..14601a83d 100644
--- a/cmd/osbuild-image-tests/constants/constants-travis.go
+++ b/cmd/osbuild-image-tests/constants/constants-travis.go
@@ -4,12 +4,12 @@ package constants
 
 import "os/exec"
 
-func GetOsbuildCommand(store string) *exec.Cmd {
+func GetOsbuildCommand(outputDirectory string) *exec.Cmd {
 	cmd := exec.Command(
 		"python3",
 		"-m", "osbuild",
 		"--libdir", ".",
-		"--store", store,
+		"--output-directory", outputDirectory,
 		"--json",
 		"-",
 	)
diff --git a/cmd/osbuild-image-tests/constants/constants.go b/cmd/osbuild-image-tests/constants/constants.go
index 73d986709..560cfc9b5 100644
--- a/cmd/osbuild-image-tests/constants/constants.go
+++ b/cmd/osbuild-image-tests/constants/constants.go
@@ -4,10 +4,10 @@ package constants
 
 import "os/exec"
 
-func GetOsbuildCommand(store string) *exec.Cmd {
+func GetOsbuildCommand(outputDirectory string) *exec.Cmd {
 	return exec.Command(
 		"osbuild",
-		"--store", store,
+		"--output-directory", outputDirectory,
 		"--json",
 		"-",
 	)
diff --git a/cmd/osbuild-image-tests/context-managers.go b/cmd/osbuild-image-tests/context-managers.go
index bf2727de3..39fede7c8 100644
--- a/cmd/osbuild-image-tests/context-managers.go
+++ b/cmd/osbuild-image-tests/context-managers.go
@@ -169,11 +169,10 @@ func withBootedQemuImage(image string, ns netNS, f func() error) error {
 
 // withBootedNspawnImage boots the specified image in the specified namespace
 // using nspawn. The VM is killed immediately after function returns.
-func withBootedNspawnImage(image, name string, ns netNS, f func() error) error {
+func withBootedNspawnImage(image string, ns netNS, f func() error) error {
 	cmd := exec.Command(
 		"systemd-nspawn",
 		"--boot", "--register=no",
-		"-M", name,
 		"--image", image,
 		"--network-namespace-path", ns.Path(),
 	)
@@ -195,11 +194,10 @@ func withBootedNspawnImage(image, name string, ns netNS, f func() error) error {
 
 // withBootedNspawnImage boots the specified directory in the specified namespace
 // using nspawn. The VM is killed immediately after function returns.
-func withBootedNspawnDirectory(dir, name string, ns netNS, f func() error) error {
+func withBootedNspawnDirectory(dir string, ns netNS, f func() error) error {
 	cmd := exec.Command(
 		"systemd-nspawn",
 		"--boot", "--register=no",
-		"-M", name,
 		"--directory", dir,
 		"--network-namespace-path", ns.Path(),
 	)
diff --git a/cmd/osbuild-image-tests/main_test.go b/cmd/osbuild-image-tests/main_test.go
index f8f88c46f..1228dbdbd 100644
--- a/cmd/osbuild-image-tests/main_test.go
+++ b/cmd/osbuild-image-tests/main_test.go
@@ -42,8 +42,8 @@ type testcaseStruct struct {
 // mutex to enforce only one osbuild instance run at a time, see below
 var osbuildMutex sync.Mutex
 
-// runOsbuild runs osbuild with the specified manifest and store.
-func runOsbuild(manifest []byte, store string) (string, error) {
+// runOsbuild runs osbuild with the specified manifest and output-directory.
+func runOsbuild(manifest []byte, outputDirectory string) error {
 	// Osbuild crashes when multiple instances are run at a time.
 	// This mutex enforces that there's always just one osbuild instance.
 	// This should be removed once osbuild is fixed.
@@ -51,7 +51,7 @@ func runOsbuild(manifest []byte, store string) (string, error) {
 	osbuildMutex.Lock()
 	defer osbuildMutex.Unlock()
 
-	cmd := constants.GetOsbuildCommand(store)
+	cmd := constants.GetOsbuildCommand(outputDirectory)
 
 	cmd.Stderr = os.Stderr
 	cmd.Stdin = bytes.NewReader(manifest)
@@ -64,21 +64,12 @@ func runOsbuild(manifest []byte, store string) (string, error) {
 		if _, ok := err.(*exec.ExitError); ok {
 			var formattedOutput bytes.Buffer
 			_ = json.Indent(&formattedOutput, outBuffer.Bytes(), "", "  ")
-			return "", fmt.Errorf("running osbuild failed: %s", formattedOutput.String())
+			return fmt.Errorf("running osbuild failed: %s", formattedOutput.String())
 		}
-		return "", fmt.Errorf("running osbuild failed from an unexpected reason: %#v", err)
+		return fmt.Errorf("running osbuild failed from an unexpected reason: %#v", err)
 	}
 
-	var result struct {
-		OutputID string `json:"output_id"`
-	}
-
-	err = json.NewDecoder(&outBuffer).Decode(&result)
-	if err != nil {
-		return "", fmt.Errorf("cannot decode osbuild output: %#v", err)
-	}
-
-	return result.OutputID, nil
+	return nil
 }
 
 // testImageInfo runs image-info on image specified by imageImage and
@@ -200,9 +191,9 @@ func testBootUsingQemu(t *testing.T, imagePath string) {
 	require.NoError(t, err)
 }
 
-func testBootUsingNspawnImage(t *testing.T, imagePath string, outputID string) {
+func testBootUsingNspawnImage(t *testing.T, imagePath string) {
 	err := withNetworkNamespace(func(ns netNS) error {
-		return withBootedNspawnImage(imagePath, outputID, ns, func() error {
+		return withBootedNspawnImage(imagePath, ns, func() error {
 			testSSH(t, "localhost", constants.TestPaths.PrivateKey, &ns)
 			return nil
 		})
@@ -210,10 +201,10 @@ func testBootUsingNspawnImage(t *testing.T, imagePath string, outputID string) {
 	require.NoError(t, err)
 }
 
-func testBootUsingNspawnDirectory(t *testing.T, imagePath string, outputID string) {
+func testBootUsingNspawnDirectory(t *testing.T, imagePath string) {
 	err := withNetworkNamespace(func(ns netNS) error {
 		return withExtractedTarArchive(imagePath, func(dir string) error {
-			return withBootedNspawnDirectory(dir, outputID, ns, func() error {
+			return withBootedNspawnDirectory(dir, ns, func() error {
 				testSSH(t, "localhost", constants.TestPaths.PrivateKey, &ns)
 				return nil
 			})
@@ -305,16 +296,16 @@ func testBootUsingAzure(t *testing.T, imagePath string) {
 // The test passes if the function is able to connect to the image via ssh
 // in defined number of attempts and systemd-is-running returns running
 // or degraded status.
-func testBoot(t *testing.T, imagePath string, bootType string, outputID string) {
+func testBoot(t *testing.T, imagePath string, bootType string) {
 	switch bootType {
 	case "qemu":
 		testBootUsingQemu(t, imagePath)
 
 	case "nspawn":
-		testBootUsingNspawnImage(t, imagePath, outputID)
+		testBootUsingNspawnImage(t, imagePath)
 
 	case "nspawn-extract":
-		testBootUsingNspawnDirectory(t, imagePath, outputID)
+		testBootUsingNspawnDirectory(t, imagePath)
 
 	case "aws":
 		testBootUsingAWS(t, imagePath)
@@ -344,7 +335,7 @@ func kvmAvailable() bool {
 
 // testImage performs a series of tests specified in the testcase
 // on an image
-func testImage(t *testing.T, testcase testcaseStruct, imagePath, outputID string) {
+func testImage(t *testing.T, testcase testcaseStruct, imagePath string) {
 	if testcase.ImageInfo != nil {
 		t.Run("image info", func(t *testing.T) {
 			testImageInfo(t, imagePath, testcase.ImageInfo)
@@ -357,7 +348,7 @@ func testImage(t *testing.T, testcase testcaseStruct, imagePath, outputID string
 			return
 		}
 		t.Run("boot", func(t *testing.T) {
-			testBoot(t, imagePath, testcase.Boot.Type, outputID)
+			testBoot(t, imagePath, testcase.Boot.Type)
 		})
 	}
 }
@@ -365,22 +356,22 @@ func testImage(t *testing.T, testcase testcaseStruct, imagePath, outputID string
 // runTestcase builds the pipeline specified in the testcase and then it
 // tests the result
 func runTestcase(t *testing.T, testcase testcaseStruct) {
-	store, err := ioutil.TempDir("/var/tmp", "osbuild-image-tests-")
-	require.NoErrorf(t, err, "cannot create temporary store: %#v", err)
+	outputDirectory, err := ioutil.TempDir("/var/tmp", "osbuild-image-tests-*")
+	require.NoErrorf(t, err, "cannot create temporary output directory: %#v", err)
 
 	defer func() {
-		err := os.RemoveAll(store)
+		err := os.RemoveAll(outputDirectory)
 		if err != nil {
-			log.Printf("cannot remove temporary store: %#v\n", err)
+			log.Printf("cannot remove temporary output directory: %#v\n", err)
 		}
 	}()
 
-	outputID, err := runOsbuild(testcase.Manifest, store)
+	err = runOsbuild(testcase.Manifest, outputDirectory)
 	require.NoError(t, err)
 
-	imagePath := fmt.Sprintf("%s/refs/%s/%s", store, outputID, testcase.ComposeRequest.Filename)
+	imagePath := fmt.Sprintf("%s/%s", outputDirectory, testcase.ComposeRequest.Filename)
 
-	testImage(t, testcase, imagePath, outputID)
+	testImage(t, testcase, imagePath)
 }
 
 // getAllCases returns paths to all testcases in the testcase directory
diff --git a/internal/jobqueue/fsjobqueue/fsjobqueue_test.go b/internal/jobqueue/fsjobqueue/fsjobqueue_test.go
index 0482557a7..108da8677 100644
--- a/internal/jobqueue/fsjobqueue/fsjobqueue_test.go
+++ b/internal/jobqueue/fsjobqueue/fsjobqueue_test.go
@@ -6,6 +6,7 @@ import (
 	"io/ioutil"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
@@ -22,11 +23,11 @@ func cleanupTempDir(t *testing.T, dir string) {
 	require.NoError(t, err)
 }
 
-func newTemporaryQueue(t *testing.T) (jobqueue.JobQueue, string) {
+func newTemporaryQueue(t *testing.T, jobTypes []string) (jobqueue.JobQueue, string) {
 	dir, err := ioutil.TempDir("", "jobqueue-test-")
 	require.NoError(t, err)
 
-	q, err := fsjobqueue.New(dir)
+	q, err := fsjobqueue.New(dir, jobTypes)
 	require.NoError(t, err)
 	require.NotNil(t, q)
 
@@ -34,6 +35,7 @@ func newTemporaryQueue(t *testing.T) (jobqueue.JobQueue, string) {
 }
 
 func pushTestJob(t *testing.T, q jobqueue.JobQueue, jobType string, args interface{}, dependencies []uuid.UUID) uuid.UUID {
+	t.Helper()
 	id, err := q.Enqueue(jobType, args, dependencies)
 	require.NoError(t, err)
 	require.NotEmpty(t, id)
@@ -52,13 +54,13 @@ func finishNextTestJob(t *testing.T, q jobqueue.JobQueue, jobType string, result
 }
 
 func TestNonExistant(t *testing.T) {
-	q, err := fsjobqueue.New("/non-existant-directory")
+	q, err := fsjobqueue.New("/non-existant-directory", []string{})
 	require.Error(t, err)
 	require.Nil(t, q)
 }
 
 func TestErrors(t *testing.T) {
-	q, dir := newTemporaryQueue(t)
+	q, dir := newTemporaryQueue(t, []string{"test"})
 	defer cleanupTempDir(t, dir)
 
 	// not serializable to JSON
@@ -78,7 +80,7 @@ func TestArgs(t *testing.T) {
 		S string
 	}
 
-	q, dir := newTemporaryQueue(t)
+	q, dir := newTemporaryQueue(t, []string{"fish", "octopus"})
 	defer cleanupTempDir(t, dir)
 
 	oneargs := argument{7, "🐠"}
@@ -100,7 +102,7 @@ func TestArgs(t *testing.T) {
 }
 
 func TestJobTypes(t *testing.T) {
-	q, dir := newTemporaryQueue(t)
+	q, dir := newTemporaryQueue(t, []string{"octopus", "clownfish"})
 	defer cleanupTempDir(t, dir)
 
 	one := pushTestJob(t, q, "octopus", nil, nil)
@@ -117,7 +119,7 @@ func TestJobTypes(t *testing.T) {
 }
 
 func TestDependencies(t *testing.T) {
-	q, dir := newTemporaryQueue(t)
+	q, dir := newTemporaryQueue(t, []string{"test"})
 	defer cleanupTempDir(t, dir)
 
 	t.Run("done-before-pushing-dependant", func(t *testing.T) {
@@ -130,15 +132,21 @@ func TestDependencies(t *testing.T) {
 		require.ElementsMatch(t, []uuid.UUID{one, two}, r)
 
 		j := pushTestJob(t, q, "test", nil, []uuid.UUID{one, two})
-		status, _, _, _, err := q.JobStatus(j, nil)
+		queued, started, finished, canceled, err := q.JobStatus(j, nil)
 		require.NoError(t, err)
-		require.Equal(t, jobqueue.JobPending, status)
+		require.True(t, !queued.IsZero())
+		require.True(t, started.IsZero())
+		require.True(t, finished.IsZero())
+		require.False(t, canceled)
 
 		require.Equal(t, j, finishNextTestJob(t, q, "test", testResult{}))
 
-		status, _, _, _, err = q.JobStatus(j, &testResult{})
+		queued, started, finished, canceled, err = q.JobStatus(j, &testResult{})
 		require.NoError(t, err)
-		require.Equal(t, jobqueue.JobFinished, status)
+		require.True(t, !queued.IsZero())
+		require.True(t, !started.IsZero())
+		require.True(t, !finished.IsZero())
+		require.False(t, canceled)
 	})
 
 	t.Run("done-after-pushing-dependant", func(t *testing.T) {
@@ -146,9 +154,12 @@ func TestDependencies(t *testing.T) {
 		two := pushTestJob(t, q, "test", nil, nil)
 
 		j := pushTestJob(t, q, "test", nil, []uuid.UUID{one, two})
-		status, _, _, _, err := q.JobStatus(j, nil)
+		queued, started, finished, canceled, err := q.JobStatus(j, nil)
 		require.NoError(t, err)
-		require.Equal(t, jobqueue.JobPending, status)
+		require.True(t, !queued.IsZero())
+		require.True(t, started.IsZero())
+		require.True(t, finished.IsZero())
+		require.False(t, canceled)
 
 		r := []uuid.UUID{}
 		r = append(r, finishNextTestJob(t, q, "test", testResult{}))
@@ -157,8 +168,90 @@ func TestDependencies(t *testing.T) {
 
 		require.Equal(t, j, finishNextTestJob(t, q, "test", testResult{}))
 
-		status, _, _, _, err = q.JobStatus(j, &testResult{})
+		queued, started, finished, canceled, err = q.JobStatus(j, &testResult{})
 		require.NoError(t, err)
-		require.Equal(t, jobqueue.JobFinished, status)
+		require.True(t, !queued.IsZero())
+		require.True(t, !started.IsZero())
+		require.True(t, !finished.IsZero())
+		require.False(t, canceled)
 	})
 }
+
+// Test that a job queue allows parallel access to multiple workers, mainly to
+// verify the quirky unlocking in Dequeue().
+func TestMultipleWorkers(t *testing.T) {
+	q, dir := newTemporaryQueue(t, []string{"octopus", "clownfish"})
+	defer cleanupTempDir(t, dir)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+		id, err := q.Dequeue(ctx, []string{"octopus"}, &json.RawMessage{})
+		require.NoError(t, err)
+		require.NotEmpty(t, id)
+	}()
+
+	// Increase the likelihood that the above goroutine was scheduled and
+	// is waiting in Dequeue().
+	time.Sleep(10 * time.Millisecond)
+
+	// This call to Dequeue() should not block on the one in the goroutine.
+	id := pushTestJob(t, q, "clownfish", nil, nil)
+	r, err := q.Dequeue(context.Background(), []string{"clownfish"}, &json.RawMessage{})
+	require.NoError(t, err)
+	require.Equal(t, id, r)
+
+	// Now wake up the Dequeue() in the goroutine and wait for it to finish.
+	_ = pushTestJob(t, q, "octopus", nil, nil)
+	<-done
+}
+
+func TestCancel(t *testing.T) {
+	q, dir := newTemporaryQueue(t, []string{"octopus", "clownfish"})
+	defer cleanupTempDir(t, dir)
+
+	// Cancel a non-existing job
+	err := q.CancelJob(uuid.New())
+	require.Error(t, err)
+
+	// Cancel a pending job
+	id := pushTestJob(t, q, "clownfish", nil, nil)
+	require.NotEmpty(t, id)
+	err = q.CancelJob(id)
+	require.NoError(t, err)
+	_, _, _, canceled, err := q.JobStatus(id, &testResult{})
+	require.NoError(t, err)
+	require.True(t, canceled)
+	err = q.FinishJob(id, &testResult{})
+	require.Error(t, err)
+
+	// Cancel a running job, which should not dequeue the canceled job from above
+	id = pushTestJob(t, q, "clownfish", nil, nil)
+	require.NotEmpty(t, id)
+	r, err := q.Dequeue(context.Background(), []string{"clownfish"}, &json.RawMessage{})
+	require.NoError(t, err)
+	require.Equal(t, id, r)
+	err = q.CancelJob(id)
+	require.NoError(t, err)
+	_, _, _, canceled, err = q.JobStatus(id, &testResult{})
+	require.NoError(t, err)
+	require.True(t, canceled)
+	err = q.FinishJob(id, &testResult{})
+	require.Error(t, err)
+
+	// Cancel a finished job, which is a no-op
+	id = pushTestJob(t, q, "clownfish", nil, nil)
+	require.NotEmpty(t, id)
+	r, err = q.Dequeue(context.Background(), []string{"clownfish"}, &json.RawMessage{})
+	require.NoError(t, err)
+	require.Equal(t, id, r)
+	err = q.FinishJob(id, &testResult{})
+	require.NoError(t, err)
+	err = q.CancelJob(id)
+	require.NoError(t, err)
+	_, _, _, canceled, err = q.JobStatus(id, &testResult{})
+	require.NoError(t, err)
+	require.False(t, canceled)
+}
diff --git a/internal/jobqueue/testjobqueue/testjobqueue.go b/internal/jobqueue/testjobqueue/testjobqueue.go
index a47afe9a1..2c0d5e3f1 100644
--- a/internal/jobqueue/testjobqueue/testjobqueue.go
+++ b/internal/jobqueue/testjobqueue/testjobqueue.go
@@ -30,7 +30,10 @@ type job struct {
 	Args         json.RawMessage
 	Dependencies []uuid.UUID
 	Result       json.RawMessage
-	Status       jobqueue.JobStatus
+	QueuedAt     time.Time
+	StartedAt    time.Time
+	FinishedAt   time.Time
+	Canceled     bool
 }
 
 func New() *testJobQueue {
@@ -45,7 +48,7 @@ func (q *testJobQueue) Enqueue(jobType string, args interface{}, dependencies []
 		Id:           uuid.New(),
 		Type:         jobType,
 		Dependencies: uniqueUUIDList(dependencies),
-		Status:       jobqueue.JobPending,
+		QueuedAt:     time.Now(),
 	}
 
 	var err error
@@ -92,7 +95,7 @@ func (q *testJobQueue) Dequeue(ctx context.Context, jobTypes []string, args inte
 			return uuid.Nil, err
 		}
 
-		j.Status = jobqueue.JobRunning
+		j.StartedAt = time.Now()
 		return j.Id, nil
 	}
 
@@ -105,7 +108,7 @@ func (q *testJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
 		return jobqueue.ErrNotExist
 	}
 
-	if j.Status != jobqueue.JobRunning {
+	if j.StartedAt.IsZero() || !j.FinishedAt.IsZero() {
 		return jobqueue.ErrNotRunning
 	}
 
@@ -115,7 +118,7 @@ func (q *testJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
 		return fmt.Errorf("error marshaling result: %v", err)
 	}
 
-	j.Status = jobqueue.JobFinished
+	j.FinishedAt = time.Now()
 
 	for _, depid := range q.dependants[id] {
 		dep := q.jobs[depid]
@@ -132,25 +135,35 @@ func (q *testJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
 	return nil
 }
 
-func (q *testJobQueue) JobStatus(id uuid.UUID, result interface{}) (status jobqueue.JobStatus, queued, started, finished time.Time, err error) {
-	var j *job
+func (q *testJobQueue) CancelJob(id uuid.UUID) error {
+	j, exists := q.jobs[id]
+	if !exists {
+		return jobqueue.ErrNotExist
+	}
+
+	j.Canceled = true
+
+	return nil
+}
 
+func (q *testJobQueue) JobStatus(id uuid.UUID, result interface{}) (queued, started, finished time.Time, canceled bool, err error) {
 	j, exists := q.jobs[id]
 	if !exists {
 		err = jobqueue.ErrNotExist
 		return
 	}
 
-	if j.Status == jobqueue.JobFinished {
+	if !j.FinishedAt.IsZero() {
 		err = json.Unmarshal(j.Result, result)
 		if err != nil {
 			return
 		}
 	}
 
-	queued = time.Time{}
-	started = time.Time{}
-	finished = time.Time{}
+	queued = j.QueuedAt
+	started = j.StartedAt
+	finished = j.FinishedAt
+	canceled = j.Canceled
 
 	return
 }
diff --git a/internal/jsondb/db_test.go b/internal/jsondb/db_test.go
index 70dbbd0bc..c18b45949 100644
--- a/internal/jsondb/db_test.go
+++ b/internal/jsondb/db_test.go
@@ -72,6 +72,39 @@ func TestCorrupt(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestRead(t *testing.T) {
+	dir, err := ioutil.TempDir("", "jsondb-test-")
+	require.NoError(t, err)
+	defer cleanupTempDir(t, dir)
+
+	err = ioutil.WriteFile(path.Join(dir, "one.json"), []byte("true"), 0755)
+	require.NoError(t, err)
+
+	db := jsondb.New(dir, 0755)
+
+	var b bool
+	exists, err := db.Read("one", &b)
+	require.NoError(t, err)
+	require.True(t, exists)
+	require.True(t, b)
+
+	// nil means don't deserialize
+	exists, err = db.Read("one", nil)
+	require.NoError(t, err)
+	require.True(t, exists)
+
+	b = false
+	exists, err = db.Read("two", &b)
+	require.NoError(t, err)
+	require.False(t, exists)
+	require.False(t, b)
+
+	// nil means don't deserialize
+	exists, err = db.Read("two", nil)
+	require.NoError(t, err)
+	require.False(t, exists)
+}
+
 func TestMultiple(t *testing.T) {
 	dir, err := ioutil.TempDir("", "jsondb-test-")
 	require.NoError(t, err)
diff --git a/tools/test-case-generators/generate-test-case b/tools/test-case-generators/generate-test-case
index ff9647b75..a7e83b3e7 100755
--- a/tools/test-case-generators/generate-test-case
+++ b/tools/test-case-generators/generate-test-case
@@ -5,6 +5,7 @@ import subprocess
 import json
 import os
 import sys
+import tempfile
 
 '''
 This script generates a json test case. It accepts a test_case_request as input through standard input.
@@ -35,10 +36,14 @@ def get_subprocess_stdout(*args, **kwargs):
     return sp.stdout
 
 
-def run_osbuild(manifest, store):
-    osbuild_cmd = ["osbuild", "--store", store, "--json", "-"]
-    result = json.loads(get_subprocess_stdout(osbuild_cmd, encoding="utf-8", input=json.dumps(manifest)))
-    return result.get("output_id")
+def run_osbuild(manifest, store, output):
+    subprocess.run(["osbuild",
+                    "--store", store,
+                    "--output-directory", output,
+                    "--json", "-"],
+                   check=True,
+                   encoding="utf-8",
+                   input=json.dumps(manifest))
 
 
 def main(test_case, store):
@@ -52,9 +57,10 @@ def main(test_case, store):
     test_case["rpmmd"] = json.loads(get_subprocess_stdout(pipeline_command, input=compose_request, encoding="utf-8"))
 
     if boot_type != "nspawn-extract":
-        output_id = run_osbuild(test_case["manifest"], store)
-        image_file = os.path.join(store, "refs", output_id, test_case["compose-request"]["filename"])
-        test_case["image-info"] = json.loads(get_subprocess_stdout(["tools/image-info", image_file], encoding="utf-8"))
+        with tempfile.TemporaryDirectory(dir=store, prefix="test-case-output-") as output:
+            run_osbuild(test_case["manifest"], store, output)
+            image_file = os.path.join(output, test_case["compose-request"]["filename"])
+            test_case["image-info"] = json.loads(get_subprocess_stdout(["tools/image-info", image_file], encoding="utf-8"))
 
     return test_case