Repository: osbuild/osbuild-composer
Pull request #1058 (2020-11-03 20:18:57)

Prepare worker package and binary for more job types
Since we only had one job type `osbuild` so far, a lot of code has assumptions about it. This PR aims to clean this up and allow for adding additional job types in the future.

This got quite large, but I hope I have explained the individual commits well enough.

No functionality is changed by these patches.

larskarlitski: I think CI is failing because of certificates expiring, which #1057 fixes. I'll rebase once that lands.
larskarlitski: Rebased now that #1057 is in. Thanks @ondrejbudai!
larskarlitski: > The test failure is an actual bug in the PR. From `journalctl`:

Indeed, thanks! I was trying to be clever and detect whether a job supports uploading of artifacts based on which API is enabled. We enable all APIs in the tests, so even jobs scheduled through the koji API tried uploading artifacts. That failed, because jobs scheduled through that API do no carry the filename of the image to be uploaded. And it shouldn't: only the weldr API uploads back to composer.

Fix this by dropping the commit that tried to be clever. Enable artifacts for all jobs server-side, but only upload if the filename is actually set.
ondrejbudai: `api.sh` test was also broken by this PR. I pushed a fix directly to this PR (see the last commit) so we can move on with this.
larskarlitski: > `api.sh` test was also broken by this PR. I pushed a fix directly to this PR (see the last commit) so we can move on with this.

Thanks! I've rebased on `master` (with your quay.io fix) and put your commit (with attribution) in the right place.

Patch:
diff --git a/cmd/osbuild-worker/jobimpl-osbuild.go b/cmd/osbuild-worker/jobimpl-osbuild.go
new file mode 100644
index 000000000..c2b85746f
--- /dev/null
+++ b/cmd/osbuild-worker/jobimpl-osbuild.go
@@ -0,0 +1,318 @@
+package main
+
+import (
+	"crypto/tls"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"net/url"
+	"os"
+	"path"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/osbuild/osbuild-composer/internal/common"
+	"github.com/osbuild/osbuild-composer/internal/distro"
+	"github.com/osbuild/osbuild-composer/internal/osbuild"
+	"github.com/osbuild/osbuild-composer/internal/target"
+	"github.com/osbuild/osbuild-composer/internal/upload/awsupload"
+	"github.com/osbuild/osbuild-composer/internal/upload/azure"
+	"github.com/osbuild/osbuild-composer/internal/upload/koji"
+	"github.com/osbuild/osbuild-composer/internal/upload/vmware"
+	"github.com/osbuild/osbuild-composer/internal/worker"
+)
+
+type OSBuildJobImpl struct {
+	Store       string
+	KojiServers map[string]koji.GSSAPICredentials
+}
+
+func packageMetadataToSignature(pkg osbuild.RPMPackageMetadata) *string {
+	if pkg.SigGPG != "" {
+		return &pkg.SigGPG
+	} else if pkg.SigPGP != "" {
+		return &pkg.SigPGP
+	}
+	return nil
+}
+
+func osbuildStagesToRPMs(stages []osbuild.StageResult) []koji.RPM {
+	rpms := make([]koji.RPM, 0)
+	for _, stage := range stages {
+		switch metadata := stage.Metadata.(type) {
+		case *osbuild.RPMStageMetadata:
+			for _, pkg := range metadata.Packages {
+				rpms = append(rpms, koji.RPM{
+					Type:      "rpm",
+					Name:      pkg.Name,
+					Epoch:     pkg.Epoch,
+					Version:   pkg.Version,
+					Release:   pkg.Release,
+					Arch:      pkg.Arch,
+					Sigmd5:    pkg.SigMD5,
+					Signature: packageMetadataToSignature(pkg),
+				})
+			}
+		default:
+			continue
+		}
+	}
+	return rpms
+}
+
+func (impl *OSBuildJobImpl) Run(job worker.Job) error {
+	outputDirectory, err := ioutil.TempDir("/var/tmp", "osbuild-worker-*")
+	if err != nil {
+		return fmt.Errorf("error creating temporary output directory: %v", err)
+	}
+	defer func() {
+		err := os.RemoveAll(outputDirectory)
+		if err != nil {
+			log.Printf("Error removing temporary output directory (%s): %v", outputDirectory, err)
+		}
+	}()
+
+	var args worker.OSBuildJob
+	err = job.Args(&args)
+	if err != nil {
+		return err
+	}
+
+	start_time := time.Now()
+
+	osbuildOutput, err := RunOSBuild(args.Manifest, impl.Store, outputDirectory, os.Stderr)
+	if err != nil {
+		return err
+	}
+
+	end_time := time.Now()
+
+	if osbuildOutput.Success && args.ImageName != "" {
+		var f *os.File
+		imagePath := path.Join(outputDirectory, args.ImageName)
+		if args.StreamOptimized {
+			f, err = vmware.OpenAsStreamOptimizedVmdk(imagePath)
+			if err != nil {
+				return err
+			}
+		} else {
+			f, err = os.Open(imagePath)
+			if err != nil {
+				return err
+			}
+		}
+		err = job.UploadArtifact(args.ImageName, f)
+		if err != nil {
+			return err
+		}
+	}
+
+	var r []error
+
+	for _, t := range args.Targets {
+		switch options := t.Options.(type) {
+		case *target.LocalTargetOptions:
+			if !osbuildOutput.Success {
+				continue
+			}
+			var f *os.File
+			imagePath := path.Join(outputDirectory, options.Filename)
+			if options.StreamOptimized {
+				f, err = vmware.OpenAsStreamOptimizedVmdk(imagePath)
+				if err != nil {
+					r = append(r, err)
+					continue
+				}
+			} else {
+				f, err = os.Open(imagePath)
+				if err != nil {
+					r = append(r, err)
+					continue
+				}
+			}
+
+			err = job.UploadArtifact(options.Filename, f)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+		case *target.AWSTargetOptions:
+			if !osbuildOutput.Success {
+				continue
+			}
+			a, err := awsupload.New(options.Region, options.AccessKeyID, options.SecretAccessKey)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+			key := options.Key
+			if key == "" {
+				key = uuid.New().String()
+			}
+
+			_, err = a.Upload(path.Join(outputDirectory, options.Filename), options.Bucket, key)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+			/* TODO: communicate back the AMI */
+			_, err = a.Register(t.ImageName, options.Bucket, key)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+		case *target.AzureTargetOptions:
+			if !osbuildOutput.Success {
+				continue
+			}
+			credentials := azure.Credentials{
+				StorageAccount:   options.StorageAccount,
+				StorageAccessKey: options.StorageAccessKey,
+			}
+			metadata := azure.ImageMetadata{
+				ContainerName: options.Container,
+				ImageName:     t.ImageName,
+			}
+
+			const azureMaxUploadGoroutines = 4
+			err := azure.UploadImage(
+				credentials,
+				metadata,
+				path.Join(outputDirectory, options.Filename),
+				azureMaxUploadGoroutines,
+			)
+
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+		case *target.KojiTargetOptions:
+			// Koji for some reason needs TLS renegotiation enabled.
+			// Clone the default http transport and enable renegotiation.
+			transport := http.DefaultTransport.(*http.Transport).Clone()
+			transport.TLSClientConfig = &tls.Config{
+				Renegotiation: tls.RenegotiateOnceAsClient,
+			}
+
+			kojiServer, _ := url.Parse(options.Server)
+			creds, exists := impl.KojiServers[kojiServer.Hostname()]
+			if !exists {
+				r = append(r, fmt.Errorf("Koji server has not been configured: %s", kojiServer.Hostname()))
+				continue
+			}
+
+			k, err := koji.NewFromGSSAPI(options.Server, &creds, transport)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+			defer func() {
+				err := k.Logout()
+				if err != nil {
+					log.Printf("koji logout failed: %v", err)
+				}
+			}()
+
+			if !osbuildOutput.Success {
+				err = k.CGFailBuild(int(options.BuildID), options.Token)
+				if err != nil {
+					log.Printf("CGFailBuild failed: %v", err)
+				}
+				continue
+			}
+
+			f, err := os.Open(path.Join(outputDirectory, options.Filename))
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+			hash, filesize, err := k.Upload(f, options.UploadDirectory, options.KojiFilename)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+			hostOS, err := distro.GetRedHatRelease()
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+
+			build := koji.ImageBuild{
+				BuildID:   options.BuildID,
+				TaskID:    options.TaskID,
+				Name:      options.Name,
+				Version:   options.Version,
+				Release:   options.Release,
+				StartTime: start_time.Unix(),
+				EndTime:   end_time.Unix(),
+			}
+			buildRoots := []koji.BuildRoot{
+				{
+					ID: 1,
+					Host: koji.Host{
+						Os:   hostOS,
+						Arch: common.CurrentArch(),
+					},
+					ContentGenerator: koji.ContentGenerator{
+						Name:    "osbuild",
+						Version: "0", // TODO: put the correct version here
+					},
+					Container: koji.Container{
+						Type: "none",
+						Arch: common.CurrentArch(),
+					},
+					Tools: []koji.Tool{},
+					RPMs:  osbuildStagesToRPMs(osbuildOutput.Build.Stages),
+				},
+			}
+			output := []koji.Image{
+				{
+					BuildRootID:  1,
+					Filename:     options.KojiFilename,
+					FileSize:     uint64(filesize),
+					Arch:         common.CurrentArch(),
+					ChecksumType: "md5",
+					MD5:          hash,
+					Type:         "image",
+					RPMs:         osbuildStagesToRPMs(osbuildOutput.Stages),
+					Extra: koji.ImageExtra{
+						Info: koji.ImageExtraInfo{
+							Arch: "noarch",
+						},
+					},
+				},
+			}
+
+			_, err = k.CGImport(build, buildRoots, output, options.UploadDirectory, options.Token)
+			if err != nil {
+				r = append(r, err)
+				continue
+			}
+		default:
+			r = append(r, fmt.Errorf("invalid target type"))
+		}
+	}
+
+	var targetErrors []string
+	for _, err := range r {
+		targetErrors = append(targetErrors, err.Error())
+	}
+
+	err = job.Update(&worker.OSBuildJobResult{
+		Success:       osbuildOutput.Success && len(targetErrors) == 0,
+		OSBuildOutput: osbuildOutput,
+		TargetErrors:  targetErrors,
+	})
+	if err != nil {
+		return fmt.Errorf("Error reporting job result: %v", err)
+	}
+
+	return nil
+}
diff --git a/cmd/osbuild-worker/main.go b/cmd/osbuild-worker/main.go
index 0bd0f6305..c06ee8ccf 100644
--- a/cmd/osbuild-worker/main.go
+++ b/cmd/osbuild-worker/main.go
@@ -9,23 +9,13 @@ import (
 	"fmt"
 	"io/ioutil"
 	"log"
-	"net/http"
-	"net/url"
 	"os"
 	"path"
 	"time"
 
 	"github.com/BurntSushi/toml"
-	"github.com/google/uuid"
 
-	"github.com/osbuild/osbuild-composer/internal/common"
-	"github.com/osbuild/osbuild-composer/internal/distro"
-	"github.com/osbuild/osbuild-composer/internal/osbuild"
-	"github.com/osbuild/osbuild-composer/internal/target"
-	"github.com/osbuild/osbuild-composer/internal/upload/awsupload"
-	"github.com/osbuild/osbuild-composer/internal/upload/azure"
 	"github.com/osbuild/osbuild-composer/internal/upload/koji"
-	"github.com/osbuild/osbuild-composer/internal/upload/vmware"
 	"github.com/osbuild/osbuild-composer/internal/worker"
 )
 
@@ -37,6 +27,11 @@ type connectionConfig struct {
 	ClientCertFile string
 }
 
+// Represents the implementation of a job type as defined by the worker API.
+type JobImplementation interface {
+	Run(job worker.Job) error
+}
+
 func createTLSConfig(config *connectionConfig) (*tls.Config, error) {
 	caCertPEM, err := ioutil.ReadFile(config.CACertFile)
 	if err != nil {
@@ -60,312 +55,6 @@ func createTLSConfig(config *connectionConfig) (*tls.Config, error) {
 	}, nil
 }
 
-type TargetsError struct {
-	Errors []error
-}
-
-func (e *TargetsError) Error() string {
-	errString := fmt.Sprintf("%d target(s) errored:\n", len(e.Errors))
-
-	for _, err := range e.Errors {
-		errString += err.Error() + "\n"
-	}
-
-	return errString
-}
-
-func packageMetadataToSignature(pkg osbuild.RPMPackageMetadata) *string {
-	if pkg.SigGPG != "" {
-		return &pkg.SigGPG
-	} else if pkg.SigPGP != "" {
-		return &pkg.SigPGP
-	}
-	return nil
-}
-
-func osbuildStagesToRPMs(stages []osbuild.StageResult) []koji.RPM {
-	rpms := make([]koji.RPM, 0)
-	for _, stage := range stages {
-		switch metadata := stage.Metadata.(type) {
-		case *osbuild.RPMStageMetadata:
-			for _, pkg := range metadata.Packages {
-				rpms = append(rpms, koji.RPM{
-					Type:      "rpm",
-					Name:      pkg.Name,
-					Epoch:     pkg.Epoch,
-					Version:   pkg.Version,
-					Release:   pkg.Release,
-					Arch:      pkg.Arch,
-					Sigmd5:    pkg.SigMD5,
-					Signature: packageMetadataToSignature(pkg),
-				})
-			}
-		default:
-			continue
-		}
-	}
-	return rpms
-}
-
-func RunJob(job worker.Job, store string, kojiServers map[string]koji.GSSAPICredentials) (*osbuild.Result, error) {
-	outputDirectory, err := ioutil.TempDir("/var/tmp", "osbuild-worker-*")
-	if err != nil {
-		return nil, fmt.Errorf("error creating temporary output directory: %v", err)
-	}
-	defer func() {
-		err := os.RemoveAll(outputDirectory)
-		if err != nil {
-			log.Printf("Error removing temporary output directory (%s): %v", outputDirectory, err)
-		}
-	}()
-
-	manifest, targets, err := job.OSBuildArgs()
-	if err != nil {
-		return nil, err
-	}
-
-	start_time := time.Now()
-
-	result, err := RunOSBuild(manifest, store, outputDirectory, os.Stderr)
-	if err != nil {
-		return nil, err
-	}
-
-	end_time := time.Now()
-
-	var r []error
-
-	for _, t := range targets {
-		switch options := t.Options.(type) {
-		case *target.LocalTargetOptions:
-			var f *os.File
-			imagePath := path.Join(outputDirectory, options.Filename)
-			if options.StreamOptimized {
-				f, err = vmware.OpenAsStreamOptimizedVmdk(imagePath)
-				if err != nil {
-					r = append(r, err)
-					continue
-				}
-			} else {
-				f, err = os.Open(imagePath)
-				if err != nil {
-					r = append(r, err)
-					continue
-				}
-			}
-
-			err = job.UploadArtifact(options.Filename, f)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-		case *target.AWSTargetOptions:
-
-			a, err := awsupload.New(options.Region, options.AccessKeyID, options.SecretAccessKey)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-			key := options.Key
-			if key == "" {
-				key = uuid.New().String()
-			}
-
-			_, err = a.Upload(path.Join(outputDirectory, options.Filename), options.Bucket, key)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-			/* TODO: communicate back the AMI */
-			_, err = a.Register(t.ImageName, options.Bucket, key)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-		case *target.AzureTargetOptions:
-
-			credentials := azure.Credentials{
-				StorageAccount:   options.StorageAccount,
-				StorageAccessKey: options.StorageAccessKey,
-			}
-			metadata := azure.ImageMetadata{
-				ContainerName: options.Container,
-				ImageName:     t.ImageName,
-			}
-
-			const azureMaxUploadGoroutines = 4
-			err := azure.UploadImage(
-				credentials,
-				metadata,
-				path.Join(outputDirectory, options.Filename),
-				azureMaxUploadGoroutines,
-			)
-
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-		case *target.KojiTargetOptions:
-			// Koji for some reason needs TLS renegotiation enabled.
-			// Clone the default http transport and enable renegotiation.
-			transport := http.DefaultTransport.(*http.Transport).Clone()
-			transport.TLSClientConfig = &tls.Config{
-				Renegotiation: tls.RenegotiateOnceAsClient,
-			}
-
-			kojiServer, _ := url.Parse(options.Server)
-			creds, exists := kojiServers[kojiServer.Hostname()]
-			if !exists {
-				r = append(r, fmt.Errorf("Koji server has not been configured: %s", kojiServer.Hostname()))
-			}
-
-			k, err := koji.NewFromGSSAPI(options.Server, &creds, transport)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-			defer func() {
-				err := k.Logout()
-				if err != nil {
-					log.Printf("koji logout failed: %v", err)
-				}
-			}()
-
-			f, err := os.Open(path.Join(outputDirectory, options.Filename))
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-			hash, filesize, err := k.Upload(f, options.UploadDirectory, options.KojiFilename)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-			hostOS, err := distro.GetRedHatRelease()
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-
-			build := koji.ImageBuild{
-				BuildID:   options.BuildID,
-				TaskID:    options.TaskID,
-				Name:      options.Name,
-				Version:   options.Version,
-				Release:   options.Release,
-				StartTime: start_time.Unix(),
-				EndTime:   end_time.Unix(),
-			}
-			buildRoots := []koji.BuildRoot{
-				{
-					ID: 1,
-					Host: koji.Host{
-						Os:   hostOS,
-						Arch: common.CurrentArch(),
-					},
-					ContentGenerator: koji.ContentGenerator{
-						Name:    "osbuild",
-						Version: "0", // TODO: put the correct version here
-					},
-					Container: koji.Container{
-						Type: "none",
-						Arch: common.CurrentArch(),
-					},
-					Tools: []koji.Tool{},
-					RPMs:  osbuildStagesToRPMs(result.Build.Stages),
-				},
-			}
-			output := []koji.Image{
-				{
-					BuildRootID:  1,
-					Filename:     options.KojiFilename,
-					FileSize:     uint64(filesize),
-					Arch:         common.CurrentArch(),
-					ChecksumType: "md5",
-					MD5:          hash,
-					Type:         "image",
-					RPMs:         osbuildStagesToRPMs(result.Stages),
-					Extra: koji.ImageExtra{
-						Info: koji.ImageExtraInfo{
-							Arch: "noarch",
-						},
-					},
-				},
-			}
-
-			_, err = k.CGImport(build, buildRoots, output, options.UploadDirectory, options.Token)
-			if err != nil {
-				r = append(r, err)
-				continue
-			}
-		default:
-			r = append(r, fmt.Errorf("invalid target type"))
-		}
-	}
-
-	err = os.RemoveAll(outputDirectory)
-	if err != nil {
-		log.Printf("Error removing osbuild output directory (%s): %v", outputDirectory, err)
-	}
-
-	if len(r) > 0 {
-		return result, &TargetsError{r}
-	}
-
-	return result, nil
-}
-
-func FailJob(job worker.Job, kojiServers map[string]koji.GSSAPICredentials) {
-	_, targets, err := job.OSBuildArgs()
-	if err != nil {
-		panic(err)
-	}
-
-	for _, t := range targets {
-		switch options := t.Options.(type) {
-		case *target.KojiTargetOptions:
-			// Koji for some reason needs TLS renegotiation enabled.
-			// Clone the default http transport and enable renegotiation.
-			transport := http.DefaultTransport.(*http.Transport).Clone()
-			transport.TLSClientConfig = &tls.Config{
-				Renegotiation: tls.RenegotiateOnceAsClient,
-			}
-
-			kojiServer, _ := url.Parse(options.Server)
-			creds, exists := kojiServers[kojiServer.Hostname()]
-			if !exists {
-				log.Printf("Koji server has not been configured: %s", kojiServer.Hostname())
-				return
-			}
-
-			k, err := koji.NewFromGSSAPI(options.Server, &creds, transport)
-			if err != nil {
-				log.Printf("koji login failed: %v", err)
-				return
-			}
-
-			defer func() {
-				err := k.Logout()
-				if err != nil {
-					log.Printf("koji logout failed: %v", err)
-				}
-			}()
-
-			err = k.CGFailBuild(int(options.BuildID), options.Token)
-			if err != nil {
-				log.Printf("CGFailBuild failed: %v", err)
-			}
-		default:
-		}
-	}
-}
-
 // Regularly ask osbuild-composer if the compose we're currently working on was
 // canceled and exit the process if it was.
 // It would be cleaner to kill the osbuild process using (`exec.CommandContext`
@@ -376,11 +65,7 @@ func WatchJob(ctx context.Context, job worker.Job) {
 		select {
 		case <-time.After(15 * time.Second):
 			canceled, err := job.Canceled()
-			if err != nil {
-				log.Printf("Error fetching job status: %v", err)
-				os.Exit(0)
-			}
-			if canceled {
+			if err == nil && canceled {
 				log.Println("Job was canceled. Exiting.")
 				os.Exit(0)
 			}
@@ -464,58 +149,43 @@ func main() {
 		}
 	}
 
+	jobImpls := map[string]JobImplementation{
+		"osbuild": &OSBuildJobImpl{
+			Store:       store,
+			KojiServers: kojiServers,
+		},
+	}
+
+	acceptedJobTypes := []string{}
+	for jt := range jobImpls {
+		acceptedJobTypes = append(acceptedJobTypes, jt)
+	}
+
 	for {
 		fmt.Println("Waiting for a new job...")
-		job, err := client.RequestJob()
+		job, err := client.RequestJob(acceptedJobTypes)
 		if err != nil {
 			log.Fatal(err)
 		}
 
-		fmt.Printf("Running job %v\n", job.Id())
+		impl, exists := jobImpls[job.Type()]
+		if !exists {
+			log.Printf("Ignoring job with unknown type %s", job.Type())
+			continue
+		}
+
+		fmt.Printf("Running '%s' job %v\n", job.Type(), job.Id())
 
-		ctx, cancel := context.WithCancel(context.Background())
+		ctx, cancelWatcher := context.WithCancel(context.Background())
 		go WatchJob(ctx, job)
 
-		var status common.ImageBuildState
-		result, err := RunJob(job, store, kojiServers)
+		err = impl.Run(job)
+		cancelWatcher()
 		if err != nil {
-			log.Printf("  Job failed: %v", err)
-			status = common.IBFailed
-
-			// Fail the jobs in any targets that expects it
-			FailJob(job, kojiServers)
-
-			// If the error comes from osbuild, retrieve the result
-			if osbuildError, ok := err.(*OSBuildError); ok {
-				result = osbuildError.Result
-			}
-
-			// Ensure we always have a non-nil result, composer doesn't like nils.
-			// This can happen in cases when OSBuild crashes and doesn't produce
-			// a meaningful output. E.g. when the machine runs of disk space.
-			if result == nil {
-				result = &osbuild.Result{
-					Success: false,
-				}
-			}
-
-			// set the success to false on every error. This is hacky but composer
-			// currently relies only on this flag to decide whether a compose was
-			// successful. There's no different way how to inform composer that
-			// e.g. an upload fail. Therefore, this line reuses the osbuild success
-			// flag to indicate all error kinds.
-			result.Success = false
-		} else {
-			log.Printf("  🎉 Job completed successfully: %v", job.Id())
-			status = common.IBFinished
+			log.Printf("Job %s failed: %v", job.Id(), err)
+			continue
 		}
 
-		// signal to WatchJob() that it can stop watching
-		cancel()
-
-		err = job.Update(status, result)
-		if err != nil {
-			log.Fatalf("Error reporting job result: %v", err)
-		}
+		log.Printf("Job %s finished", job.Id())
 	}
 }
diff --git a/cmd/osbuild-worker/osbuild.go b/cmd/osbuild-worker/osbuild.go
index d2c19914b..6c5a6c510 100644
--- a/cmd/osbuild-worker/osbuild.go
+++ b/cmd/osbuild-worker/osbuild.go
@@ -10,15 +10,11 @@ import (
 	"github.com/osbuild/osbuild-composer/internal/osbuild"
 )
 
-type OSBuildError struct {
-	Message string
-	Result  *osbuild.Result
-}
-
-func (e *OSBuildError) Error() string {
-	return e.Message
-}
-
+// Run an instance of osbuild, returning a parsed osbuild.Result.
+//
+// Note that osbuild returns non-zero when the pipeline fails. This function
+// does not return an error in this case. Instead, the failure is communicated
+// with its corresponding logs through osbuild.Result.
 func RunOSBuild(manifest distro.Manifest, store, outputDirectory string, errorWriter io.Writer) (*osbuild.Result, error) {
 	cmd := exec.Command(
 		"osbuild",
@@ -47,8 +43,11 @@ func RunOSBuild(manifest distro.Manifest, store, outputDirectory string, errorWr
 	if err != nil {
 		return nil, fmt.Errorf("error encoding osbuild pipeline: %v", err)
 	}
-	// FIXME: handle or comment this possible error
-	_ = stdin.Close()
+
+	err = stdin.Close()
+	if err != nil {
+		return nil, fmt.Errorf("error closing osbuild's stdin: %v", err)
+	}
 
 	var result osbuild.Result
 	err = json.NewDecoder(stdout).Decode(&result)
@@ -58,9 +57,9 @@ func RunOSBuild(manifest distro.Manifest, store, outputDirectory string, errorWr
 
 	err = cmd.Wait()
 	if err != nil {
-		return nil, &OSBuildError{
-			Message: fmt.Sprintf("running osbuild failed: %v", err),
-			Result:  &result,
+		// ignore ExitError if output could be decoded correctly
+		if _, isExitError := err.(*exec.ExitError); !isExitError {
+			return nil, fmt.Errorf("running osbuild failed: %v", err)
 		}
 	}
 
diff --git a/internal/cloudapi/openapi.yml b/internal/cloudapi/openapi.yml
index e8103cb16..8a08678ed 100644
--- a/internal/cloudapi/openapi.yml
+++ b/internal/cloudapi/openapi.yml
@@ -69,7 +69,7 @@ components:
       properties:
         status:
           type: string
-          enum: ['success', 'failure', 'pending']
+          enum: ['success', 'failure', 'pending', 'running']
           example: 'success'
         image_statuses:
           type: array
diff --git a/internal/cloudapi/server.go b/internal/cloudapi/server.go
index 9b3569108..3e60438f9 100644
--- a/internal/cloudapi/server.go
+++ b/internal/cloudapi/server.go
@@ -17,6 +17,13 @@ import (
 	"github.com/osbuild/osbuild-composer/internal/worker"
 )
 
+const (
+	StatusPending = "pending"
+	StatusRunning = "running"
+	StatusSuccess = "success"
+	StatusFailure = "failure"
+)
+
 // Server represents the state of the cloud Server
 type Server struct {
 	workers     *worker.Server
@@ -181,7 +188,10 @@ func (server *Server) Compose(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	id, err := server.workers.Enqueue(ir.arch, ir.manifest, targets)
+	id, err := server.workers.Enqueue(ir.arch, &worker.OSBuildJob{
+		Manifest: ir.manifest,
+		Targets:  targets,
+	})
 	if err != nil {
 		http.Error(w, "Failed to enqueue manifest", http.StatusInternalServerError)
 		return
@@ -199,24 +209,24 @@ func (server *Server) Compose(w http.ResponseWriter, r *http.Request) {
 
 // ComposeStatus handles a /compose/{id} GET request
 func (server *Server) ComposeStatus(w http.ResponseWriter, r *http.Request, id string) {
-	composeId, err := uuid.Parse(id)
-
+	jobId, err := uuid.Parse(id)
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Invalid format for parameter id: %s", err), http.StatusBadRequest)
 		return
 	}
 
-	status, err := server.workers.JobStatus(composeId)
+	var result worker.OSBuildJobResult
+	status, err := server.workers.JobStatus(jobId, &result)
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Job %s not found: %s", id, err), http.StatusNotFound)
 		return
 	}
 
 	response := ComposeStatus{
-		Status: status.State.ToString(), // TODO: map the status correctly
+		Status: composeStatusFromJobStatus(status, &result),
 		ImageStatuses: &[]ImageStatus{
 			{
-				Status: status.State.ToString(), // TODO: map the status correctly
+				Status: composeStatusFromJobStatus(status, &result),
 			},
 		},
 	}
@@ -226,3 +236,23 @@ func (server *Server) ComposeStatus(w http.ResponseWriter, r *http.Request, id s
 		panic("Failed to write response")
 	}
 }
+
+func composeStatusFromJobStatus(js *worker.JobStatus, result *worker.OSBuildJobResult) string {
+	if js.Canceled {
+		return StatusFailure
+	}
+
+	if js.Started.IsZero() {
+		return StatusPending
+	}
+
+	if js.Finished.IsZero() {
+		return StatusRunning
+	}
+
+	if result.Success {
+		return StatusSuccess
+	}
+
+	return StatusFailure
+}
diff --git a/internal/common/states.go b/internal/common/states.go
index 0de0387d3..3352210b2 100644
--- a/internal/common/states.go
+++ b/internal/common/states.go
@@ -69,31 +69,3 @@ func (ibs *ImageBuildState) UnmarshalJSON(data []byte) error {
 func (ibs ImageBuildState) MarshalJSON() ([]byte, error) {
 	return json.Marshal(getStateMapping()[ibs])
 }
-
-type ComposeState int
-
-const (
-	CWaiting ComposeState = iota
-	CRunning
-	CFinished
-	CFailed
-)
-
-// ToString converts ImageBuildState into a human readable string
-func (cs ComposeState) ToString() string {
-	return getStateMapping()[int(cs)]
-}
-
-// UnmarshalJSON converts a JSON string into an ImageBuildState
-func (ibs *ComposeState) UnmarshalJSON(data []byte) error {
-	val, err := unmarshalStateHelper(data, getStateMapping())
-	if err != nil {
-		return err
-	}
-	*ibs = ComposeState(val)
-	return nil
-}
-
-func (ibs ComposeState) MarshalJSON() ([]byte, error) {
-	return json.Marshal(getStateMapping()[ibs])
-}
diff --git a/internal/jobqueue/fsjobqueue/fsjobqueue.go b/internal/jobqueue/fsjobqueue/fsjobqueue.go
index 7a407d95d..7949bf748 100644
--- a/internal/jobqueue/fsjobqueue/fsjobqueue.go
+++ b/internal/jobqueue/fsjobqueue/fsjobqueue.go
@@ -145,13 +145,13 @@ func (q *fsJobQueue) Enqueue(jobType string, args interface{}, dependencies []uu
 	return j.Id, nil
 }
 
-func (q *fsJobQueue) Dequeue(ctx context.Context, jobTypes []string, args interface{}) (uuid.UUID, error) {
+func (q *fsJobQueue) Dequeue(ctx context.Context, jobTypes []string) (uuid.UUID, string, json.RawMessage, error) {
 	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
+		return uuid.Nil, "", nil, err
 	}
 
 	// Filter q.pending by the `jobTypes`. Ignore those job types that this
@@ -173,12 +173,12 @@ func (q *fsJobQueue) Dequeue(ctx context.Context, jobTypes []string, args interf
 		q.mu.Lock()
 
 		if err != nil {
-			return uuid.Nil, err
+			return uuid.Nil, "", nil, err
 		}
 
 		j, err = q.readJob(id)
 		if err != nil {
-			return uuid.Nil, err
+			return uuid.Nil, "", nil, err
 		}
 
 		if !j.Canceled {
@@ -186,19 +186,14 @@ func (q *fsJobQueue) Dequeue(ctx context.Context, jobTypes []string, args interf
 		}
 	}
 
-	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.StartedAt = time.Now()
 
-	err = q.db.Write(j.Id.String(), j)
+	err := q.db.Write(j.Id.String(), j)
 	if err != nil {
-		return uuid.Nil, fmt.Errorf("error writing job %s: %v", j.Id, err)
+		return uuid.Nil, "", nil, fmt.Errorf("error writing job %s: %v", j.Id, err)
 	}
 
-	return j.Id, nil
+	return j.Id, j.Type, j.Args, nil
 }
 
 func (q *fsJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
diff --git a/internal/jobqueue/jobqueue.go b/internal/jobqueue/jobqueue.go
index 6dbd96b0e..751952c35 100644
--- a/internal/jobqueue/jobqueue.go
+++ b/internal/jobqueue/jobqueue.go
@@ -14,6 +14,7 @@ package jobqueue
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"time"
 
@@ -38,11 +39,9 @@ type JobQueue interface {
 	// Waits until a job with a type of any of `jobTypes` is available, or `ctx` is
 	// canceled.
 	//
-	// All jobs in `jobTypes` must take the same type of `args`, corresponding to
-	// the one that was passed to Enqueue().
-	//
-	// Returns the job's id or an error.
-	Dequeue(ctx context.Context, jobTypes []string, args interface{}) (uuid.UUID, error)
+	// Returns the job's id, type, and arguments, or an error. Arguments
+	// can be unmarshaled to the type given in Enqueue().
+	Dequeue(ctx context.Context, jobTypes []string) (uuid.UUID, string, json.RawMessage, error)
 
 	// Mark the job with `id` as finished. `result` must fit the associated
 	// job type and must be serializable to JSON.
diff --git a/internal/kojiapi/server.go b/internal/kojiapi/server.go
index 8666c3672..c7a029364 100644
--- a/internal/kojiapi/server.go
+++ b/internal/kojiapi/server.go
@@ -14,7 +14,6 @@ import (
 	"github.com/labstack/echo/v4"
 
 	"github.com/osbuild/osbuild-composer/internal/blueprint"
-	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/distro"
 	"github.com/osbuild/osbuild-composer/internal/kojiapi/api"
 	"github.com/osbuild/osbuild-composer/internal/rpmmd"
@@ -179,20 +178,25 @@ func (h *apiHandlers) PostCompose(ctx echo.Context) error {
 		return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Could not initialize build with koji: %v", err))
 	}
 
-	id, err := h.server.workers.Enqueue(ir.arch, ir.manifest, []*target.Target{
-		target.NewKojiTarget(&target.KojiTargetOptions{
-			BuildID:         uint64(buildInfo.BuildID),
-			TaskID:          uint64(request.Koji.TaskId),
-			Token:           buildInfo.Token,
-			Name:            request.Name,
-			Version:         request.Version,
-			Release:         request.Release,
-			Filename:        ir.filename,
-			UploadDirectory: "osbuild-composer-koji-" + uuid.New().String(),
-			Server:          request.Koji.Server,
-			KojiFilename:    ir.kojiFilename,
-		}),
-	})
+	job := worker.OSBuildJob{
+		Manifest: ir.manifest,
+		Targets: []*target.Target{
+			target.NewKojiTarget(&target.KojiTargetOptions{
+				BuildID:         uint64(buildInfo.BuildID),
+				TaskID:          uint64(request.Koji.TaskId),
+				Token:           buildInfo.Token,
+				Name:            request.Name,
+				Version:         request.Version,
+				Release:         request.Release,
+				Filename:        ir.filename,
+				UploadDirectory: "osbuild-composer-koji-" + uuid.New().String(),
+				Server:          request.Koji.Server,
+				KojiFilename:    ir.kojiFilename,
+			}),
+		},
+	}
+
+	id, err := h.server.workers.Enqueue(ir.arch, &job)
 	if err != nil {
 		// This is a programming errror.
 		panic(err)
@@ -223,34 +227,44 @@ func splitExtension(filename string) string {
 	return "." + strings.Join(filenameParts[1:], ".")
 }
 
-func composeStateToStatus(state common.ComposeState) string {
-	switch state {
-	case common.CFailed:
+func composeStatusFromJobStatus(js *worker.JobStatus, result *worker.OSBuildJobResult) string {
+	if js.Canceled {
 		return "failure"
-	case common.CFinished:
-		return "success"
-	case common.CRunning:
+	}
+
+	if js.Started.IsZero() {
 		return "pending"
-	case common.CWaiting:
+	}
+
+	if js.Finished.IsZero() {
 		return "pending"
-	default:
-		panic("invalid compose state")
 	}
+
+	if result.Success {
+		return "success"
+	}
+
+	return "failure"
 }
 
-func composeStateToImageStatus(state common.ComposeState) string {
-	switch state {
-	case common.CFailed:
+func imageStatusFromJobStatus(js *worker.JobStatus, result *worker.OSBuildJobResult) string {
+	if js.Canceled {
 		return "failure"
-	case common.CFinished:
-		return "success"
-	case common.CRunning:
-		return "building"
-	case common.CWaiting:
+	}
+
+	if js.Started.IsZero() {
 		return "pending"
-	default:
-		panic("invalid compose state")
 	}
+
+	if js.Finished.IsZero() {
+		return "building"
+	}
+
+	if result.Success {
+		return "success"
+	}
+
+	return "failure"
 }
 
 // GetComposeId handles a /compose/{id} GET request
@@ -260,16 +274,17 @@ func (h *apiHandlers) GetComposeId(ctx echo.Context, idstr string) error {
 		return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Invalid format for parameter id: %s", err))
 	}
 
-	status, err := h.server.workers.JobStatus(id)
+	var result worker.OSBuildJobResult
+	status, err := h.server.workers.JobStatus(id, &result)
 	if err != nil {
 		return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Job %s not found: %s", idstr, err))
 	}
 
 	response := api.ComposeStatus{
-		Status: composeStateToStatus(status.State),
+		Status: composeStatusFromJobStatus(status, &result),
 		ImageStatuses: []api.ImageStatus{
 			{
-				Status: composeStateToImageStatus(status.State),
+				Status: imageStatusFromJobStatus(status, &result),
 			},
 		},
 	}
diff --git a/internal/weldr/api.go b/internal/weldr/api.go
index e1edcb606..6359ef474 100644
--- a/internal/weldr/api.go
+++ b/internal/weldr/api.go
@@ -49,6 +49,31 @@ type API struct {
 	compatOutputDir string
 }
 
+type ComposeState int
+
+const (
+	ComposeWaiting ComposeState = iota
+	ComposeRunning
+	ComposeFinished
+	ComposeFailed
+)
+
+// ToString converts ImageBuildState into a human readable string
+func (cs ComposeState) ToString() string {
+	switch cs {
+	case ComposeWaiting:
+		return "WAITING"
+	case ComposeRunning:
+		return "RUNNING"
+	case ComposeFinished:
+		return "FINISHED"
+	case ComposeFailed:
+		return "FAILED"
+	default:
+		panic("invalid ComposeState value")
+	}
+}
+
 // systemRepoIDs returns a list of the system repos
 // NOTE: The system repos have no concept of id vs. name so the id is returned
 func (api *API) systemRepoNames() (names []string) {
@@ -162,13 +187,33 @@ func (api *API) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 }
 
 type composeStatus struct {
-	State    common.ComposeState
+	State    ComposeState
 	Queued   time.Time
 	Started  time.Time
 	Finished time.Time
 	Result   *osbuild.Result
 }
 
+func composeStateFromJobStatus(js *worker.JobStatus, result *worker.OSBuildJobResult) ComposeState {
+	if js.Canceled {
+		return ComposeFailed
+	}
+
+	if js.Started.IsZero() {
+		return ComposeWaiting
+	}
+
+	if js.Finished.IsZero() {
+		return ComposeRunning
+	}
+
+	if result.Success {
+		return ComposeFinished
+	}
+
+	return ComposeFailed
+}
+
 // Returns the state of the image in `compose` and the times the job was
 // queued, started, and finished. Assumes that there's only one image in the
 // compose.
@@ -179,16 +224,16 @@ func (api *API) getComposeStatus(compose store.Compose) *composeStatus {
 	// the job queue from the store still contain their valid status and
 	// times. Return those here as a fallback.
 	if jobId == uuid.Nil {
-		var state common.ComposeState
+		var state ComposeState
 		switch compose.ImageBuild.QueueStatus {
 		case common.IBWaiting:
-			state = common.CWaiting
+			state = ComposeWaiting
 		case common.IBRunning:
-			state = common.CRunning
+			state = ComposeRunning
 		case common.IBFinished:
-			state = common.CFinished
+			state = ComposeFinished
 		case common.IBFailed:
-			state = common.CFailed
+			state = ComposeFailed
 		}
 		return &composeStatus{
 			State:    state,
@@ -199,14 +244,17 @@ func (api *API) getComposeStatus(compose store.Compose) *composeStatus {
 		}
 	}
 
+	// All jobs are "osbuild" jobs.
+	var result worker.OSBuildJobResult
+
 	// is it ok to ignore this error?
-	jobStatus, _ := api.workers.JobStatus(jobId)
+	jobStatus, _ := api.workers.JobStatus(jobId, &result)
 	return &composeStatus{
-		State:    jobStatus.State,
+		State:    composeStateFromJobStatus(jobStatus, &result),
 		Queued:   jobStatus.Queued,
 		Started:  jobStatus.Started,
 		Finished: jobStatus.Finished,
-		Result:   jobStatus.Result.OSBuildOutput,
+		Result:   result.OSBuildOutput,
 	}
 }
 
@@ -1754,15 +1802,6 @@ func (api *API) composeHandler(writer http.ResponseWriter, request *http.Request
 		targets = append(targets, t)
 	}
 
-	targets = append(targets, target.NewLocalTarget(
-		&target.LocalTargetOptions{
-			ComposeId:       composeID,
-			ImageBuildId:    0,
-			Filename:        imageType.Filename(),
-			StreamOptimized: imageType.Name() == "vmdk", // TODO: move conversion to osbuild
-		},
-	))
-
 	bp := api.store.GetBlueprintCommitted(cr.BlueprintName)
 	if bp == nil {
 		errors := responseError{
@@ -1825,7 +1864,12 @@ func (api *API) composeHandler(writer http.ResponseWriter, request *http.Request
 	} else {
 		var jobId uuid.UUID
 
-		jobId, err = api.workers.Enqueue(api.arch.Name(), manifest, targets)
+		jobId, err = api.workers.Enqueue(api.arch.Name(), &worker.OSBuildJob{
+			Manifest:        manifest,
+			Targets:         targets,
+			ImageName:       imageType.Filename(),
+			StreamOptimized: imageType.Name() == "vmdk", // https://github.com/osbuild/osbuild/issues/528
+		})
 		if err == nil {
 			err = api.store.PushCompose(composeID, manifest, imageType, bp, size, targets, jobId)
 		}
@@ -1890,7 +1934,7 @@ func (api *API) composeDeleteHandler(writer http.ResponseWriter, request *http.R
 		}
 
 		composeStatus := api.getComposeStatus(compose)
-		if composeStatus.State != common.CFinished && composeStatus.State != common.CFailed {
+		if composeStatus.State != ComposeFinished && composeStatus.State != ComposeFailed {
 			errors = append(errors, composeDeleteError{
 				"BuildInWrongState",
 				fmt.Sprintf("Compose %s is not in FINISHED or FAILED.", id),
@@ -1955,7 +1999,7 @@ func (api *API) composeCancelHandler(writer http.ResponseWriter, request *http.R
 	}
 
 	composeStatus := api.getComposeStatus(compose)
-	if composeStatus.State == common.CWaiting {
+	if composeStatus.State == ComposeWaiting {
 		errors := responseError{
 			ID:  "BuildInWrongState",
 			Msg: fmt.Sprintf("Build %s has not started yet. No logs to view.", uuidString),
@@ -2015,9 +2059,9 @@ func (api *API) composeQueueHandler(writer http.ResponseWriter, request *http.Re
 	for id, compose := range composes {
 		composeStatus := api.getComposeStatus(compose)
 		switch composeStatus.State {
-		case common.CWaiting:
+		case ComposeWaiting:
 			reply.New = append(reply.New, composeToComposeEntry(id, compose, composeStatus, includeUploads))
-		case common.CRunning:
+		case ComposeRunning:
 			reply.Run = append(reply.Run, composeToComposeEntry(id, compose, composeStatus, includeUploads))
 		}
 	}
@@ -2199,7 +2243,7 @@ func (api *API) composeImageHandler(writer http.ResponseWriter, request *http.Re
 	}
 
 	composeStatus := api.getComposeStatus(compose)
-	if composeStatus.State != common.CFinished {
+	if composeStatus.State != ComposeFinished {
 		errors := responseError{
 			ID:  "BuildInWrongState",
 			Msg: fmt.Sprintf("Build %s is in wrong state: %s", uuidString, composeStatus.State.ToString()),
@@ -2257,7 +2301,7 @@ func (api *API) composeMetadataHandler(writer http.ResponseWriter, request *http
 	}
 
 	composeStatus := api.getComposeStatus(compose)
-	if composeStatus.State != common.CFinished && composeStatus.State != common.CFailed {
+	if composeStatus.State != ComposeFinished && composeStatus.State != ComposeFailed {
 		errors := responseError{
 			ID:  "BuildInWrongState",
 			Msg: fmt.Sprintf("Build %s is in wrong state: %s", uuidString, composeStatus.State.ToString()),
@@ -2318,7 +2362,7 @@ func (api *API) composeResultsHandler(writer http.ResponseWriter, request *http.
 	}
 
 	composeStatus := api.getComposeStatus(compose)
-	if composeStatus.State != common.CFinished && composeStatus.State != common.CFailed {
+	if composeStatus.State != ComposeFinished && composeStatus.State != ComposeFailed {
 		errors := responseError{
 			ID:  "BuildInWrongState",
 			Msg: fmt.Sprintf("Build %s is in wrong state: %s", uuidString, composeStatus.State.ToString()),
@@ -2409,7 +2453,7 @@ func (api *API) composeLogsHandler(writer http.ResponseWriter, request *http.Req
 	}
 
 	composeStatus := api.getComposeStatus(compose)
-	if composeStatus.State != common.CFinished && composeStatus.State != common.CFailed {
+	if composeStatus.State != ComposeFinished && composeStatus.State != ComposeFailed {
 		errors := responseError{
 			ID:  "BuildInWrongState",
 			Msg: fmt.Sprintf("Build %s not in FINISHED or FAILED state.", uuidString),
@@ -2473,7 +2517,7 @@ func (api *API) composeLogHandler(writer http.ResponseWriter, request *http.Requ
 	}
 
 	composeStatus := api.getComposeStatus(compose)
-	if composeStatus.State == common.CWaiting {
+	if composeStatus.State == ComposeWaiting {
 		errors := responseError{
 			ID:  "BuildInWrongState",
 			Msg: fmt.Sprintf("Build %s has not started yet. No logs to view.", uuidString),
@@ -2482,7 +2526,7 @@ func (api *API) composeLogHandler(writer http.ResponseWriter, request *http.Requ
 		return
 	}
 
-	if composeStatus.State == common.CRunning {
+	if composeStatus.State == ComposeRunning {
 		fmt.Fprintf(writer, "Build %s is still running.\n", uuidString)
 		return
 	}
@@ -2503,7 +2547,7 @@ func (api *API) composeFinishedHandler(writer http.ResponseWriter, request *http
 	includeUploads := isRequestVersionAtLeast(params, 1)
 	for id, compose := range api.store.GetAllComposes() {
 		composeStatus := api.getComposeStatus(compose)
-		if composeStatus.State != common.CFinished {
+		if composeStatus.State != ComposeFinished {
 			continue
 		}
 		reply.Finished = append(reply.Finished, composeToComposeEntry(id, compose, composeStatus, includeUploads))
@@ -2526,7 +2570,7 @@ func (api *API) composeFailedHandler(writer http.ResponseWriter, request *http.R
 	includeUploads := isRequestVersionAtLeast(params, 1)
 	for id, compose := range api.store.GetAllComposes() {
 		composeStatus := api.getComposeStatus(compose)
-		if composeStatus.State != common.CFailed {
+		if composeStatus.State != ComposeFailed {
 			continue
 		}
 		reply.Failed = append(reply.Failed, composeToComposeEntry(id, compose, composeStatus, includeUploads))
diff --git a/internal/weldr/compose.go b/internal/weldr/compose.go
index dfc90e714..5aa4f27a4 100644
--- a/internal/weldr/compose.go
+++ b/internal/weldr/compose.go
@@ -35,23 +35,23 @@ func composeToComposeEntry(id uuid.UUID, compose store.Compose, status *composeS
 	}
 
 	switch status.State {
-	case common.CWaiting:
+	case ComposeWaiting:
 		composeEntry.QueueStatus = common.IBWaiting
 		composeEntry.JobCreated = float64(status.Queued.UnixNano()) / 1000000000
 
-	case common.CRunning:
+	case ComposeRunning:
 		composeEntry.QueueStatus = common.IBRunning
 		composeEntry.JobCreated = float64(status.Queued.UnixNano()) / 1000000000
 		composeEntry.JobStarted = float64(status.Started.UnixNano()) / 1000000000
 
-	case common.CFinished:
+	case ComposeFinished:
 		composeEntry.QueueStatus = common.IBFinished
 		composeEntry.ImageSize = compose.ImageBuild.Size
 		composeEntry.JobCreated = float64(status.Queued.UnixNano()) / 1000000000
 		composeEntry.JobStarted = float64(status.Started.UnixNano()) / 1000000000
 		composeEntry.JobFinished = float64(status.Finished.UnixNano()) / 1000000000
 
-	case common.CFailed:
+	case ComposeFailed:
 		composeEntry.QueueStatus = common.IBFailed
 		composeEntry.JobCreated = float64(status.Queued.UnixNano()) / 1000000000
 		composeEntry.JobStarted = float64(status.Started.UnixNano()) / 1000000000
diff --git a/internal/weldr/upload.go b/internal/weldr/upload.go
index 4836ab375..6c2e55f1d 100644
--- a/internal/weldr/upload.go
+++ b/internal/weldr/upload.go
@@ -91,7 +91,7 @@ func (u *uploadRequest) UnmarshalJSON(data []byte) error {
 //
 // This also ignores any sensitive data passed into targets. Access keys may
 // be passed as input to composer, but should not be possible to be queried.
-func targetsToUploadResponses(targets []*target.Target, state common.ComposeState) []uploadResponse {
+func targetsToUploadResponses(targets []*target.Target, state ComposeState) []uploadResponse {
 	var uploads []uploadResponse
 	for _, t := range targets {
 		upload := uploadResponse{
@@ -101,13 +101,13 @@ func targetsToUploadResponses(targets []*target.Target, state common.ComposeStat
 		}
 
 		switch state {
-		case common.CWaiting:
+		case ComposeWaiting:
 			upload.Status = common.IBWaiting
-		case common.CRunning:
+		case ComposeRunning:
 			upload.Status = common.IBRunning
-		case common.CFinished:
+		case ComposeFinished:
 			upload.Status = common.IBFinished
-		case common.CFailed:
+		case ComposeFailed:
 			upload.Status = common.IBFailed
 		}
 
diff --git a/internal/worker/client.go b/internal/worker/client.go
index 916d61a3b..09dbe6e0e 100644
--- a/internal/worker/client.go
+++ b/internal/worker/client.go
@@ -5,7 +5,6 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -14,9 +13,6 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/osbuild/osbuild-composer/internal/common"
-	"github.com/osbuild/osbuild-composer/internal/distro"
-	"github.com/osbuild/osbuild-composer/internal/osbuild"
-	"github.com/osbuild/osbuild-composer/internal/target"
 	"github.com/osbuild/osbuild-composer/internal/worker/api"
 )
 
@@ -27,8 +23,9 @@ type Client struct {
 
 type Job interface {
 	Id() uuid.UUID
-	OSBuildArgs() (distro.Manifest, []*target.Target, error)
-	Update(status common.ImageBuildState, result *osbuild.Result) error
+	Type() string
+	Args(args interface{}) error
+	Update(result interface{}) error
 	Canceled() (bool, error)
 	UploadArtifact(name string, reader io.Reader) error
 }
@@ -84,7 +81,7 @@ func NewClientUnix(path string) *Client {
 	return &Client{server, requester}
 }
 
-func (c *Client) RequestJob() (Job, error) {
+func (c *Client) RequestJob(types []string) (Job, error) {
 	url, err := c.server.Parse("jobs")
 	if err != nil {
 		// This only happens when "jobs" cannot be parsed.
@@ -93,7 +90,7 @@ func (c *Client) RequestJob() (Job, error) {
 
 	var buf bytes.Buffer
 	err = json.NewEncoder(&buf).Encode(api.RequestJobJSONRequestBody{
-		Types: []string{"osbuild"},
+		Types: types,
 		Arch:  common.CurrentArch(),
 	})
 	if err != nil {
@@ -140,25 +137,22 @@ func (j *job) Id() uuid.UUID {
 	return j.id
 }
 
-func (j *job) OSBuildArgs() (distro.Manifest, []*target.Target, error) {
-	if j.jobType != "osbuild" {
-		return nil, nil, errors.New("not an osbuild job")
-	}
+func (j *job) Type() string {
+	return j.jobType
+}
 
-	var args OSBuildJob
-	err := json.Unmarshal(j.args, &args)
+func (j *job) Args(args interface{}) error {
+	err := json.Unmarshal(j.args, args)
 	if err != nil {
-		return nil, nil, fmt.Errorf("error parsing osbuild job arguments: %v", err)
+		return fmt.Errorf("error parsing job arguments: %v", err)
 	}
-
-	return args.Manifest, args.Targets, nil
+	return nil
 }
 
-func (j *job) Update(status common.ImageBuildState, result *osbuild.Result) error {
+func (j *job) Update(result interface{}) error {
 	var buf bytes.Buffer
 	err := json.NewEncoder(&buf).Encode(api.UpdateJobJSONRequestBody{
 		Result: result,
-		Status: status.ToString(),
 	})
 	if err != nil {
 		panic(err)
diff --git a/internal/worker/json.go b/internal/worker/json.go
index 977adc6a4..b7bfcb348 100644
--- a/internal/worker/json.go
+++ b/internal/worker/json.go
@@ -4,7 +4,6 @@ import (
 	"encoding/json"
 
 	"github.com/google/uuid"
-	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/distro"
 	"github.com/osbuild/osbuild-composer/internal/osbuild"
 	"github.com/osbuild/osbuild-composer/internal/target"
@@ -15,12 +14,16 @@ import (
 //
 
 type OSBuildJob struct {
-	Manifest distro.Manifest  `json:"manifest"`
-	Targets  []*target.Target `json:"targets,omitempty"`
+	Manifest        distro.Manifest  `json:"manifest"`
+	Targets         []*target.Target `json:"targets,omitempty"`
+	ImageName       string           `json:"image_name,omitempty"`
+	StreamOptimized bool             `json:"stream_optimized,omitempty"`
 }
 
 type OSBuildJobResult struct {
+	Success       bool            `json:"success"`
 	OSBuildOutput *osbuild.Result `json:"osbuild_output,omitempty"`
+	TargetErrors  []string        `json:"target_errors,omitempty"`
 }
 
 //
@@ -44,8 +47,7 @@ type getJobResponse struct {
 }
 
 type updateJobRequest struct {
-	Status common.ImageBuildState `json:"status"`
-	Result *osbuild.Result        `json:"result"`
+	Result json.RawMessage `json:"result"`
 }
 
 type updateJobResponse struct {
diff --git a/internal/worker/server.go b/internal/worker/server.go
index ec81739a2..7a8fc8432 100644
--- a/internal/worker/server.go
+++ b/internal/worker/server.go
@@ -18,10 +18,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/labstack/echo/v4"
 
-	"github.com/osbuild/osbuild-composer/internal/common"
-	"github.com/osbuild/osbuild-composer/internal/distro"
 	"github.com/osbuild/osbuild-composer/internal/jobqueue"
-	"github.com/osbuild/osbuild-composer/internal/target"
 	"github.com/osbuild/osbuild-composer/internal/worker/api"
 )
 
@@ -44,12 +41,10 @@ type Server struct {
 }
 
 type JobStatus struct {
-	State    common.ComposeState
 	Queued   time.Time
 	Started  time.Time
 	Finished time.Time
 	Canceled bool
-	Result   OSBuildJobResult
 }
 
 var ErrTokenNotExist = errors.New("worker token does not exist")
@@ -88,43 +83,29 @@ func (s *Server) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 	s.server.Handler.ServeHTTP(writer, request)
 }
 
-func (s *Server) Enqueue(arch string, manifest distro.Manifest, targets []*target.Target) (uuid.UUID, error) {
-	job := OSBuildJob{
-		Manifest: manifest,
-		Targets:  targets,
-	}
-
+func (s *Server) Enqueue(arch string, job *OSBuildJob) (uuid.UUID, error) {
 	return s.jobs.Enqueue("osbuild:"+arch, job, nil)
 }
 
-func (s *Server) JobStatus(id uuid.UUID) (*JobStatus, error) {
-	var canceled bool
-	var result OSBuildJobResult
-
-	queued, started, finished, canceled, err := s.jobs.JobStatus(id, &result)
+func (s *Server) JobStatus(id uuid.UUID, result interface{}) (*JobStatus, error) {
+	queued, started, finished, canceled, err := s.jobs.JobStatus(id, result)
 	if err != nil {
 		return nil, err
 	}
-	state := common.CWaiting
-	if canceled {
-		state = common.CFailed
-	} else if !finished.IsZero() {
-		if result.OSBuildOutput != nil && result.OSBuildOutput.Success {
-			state = common.CFinished
-		} else {
-			state = common.CFailed
+
+	// For backwards compatibility: OSBuildJobResult didn't use to have a
+	// top-level `Success` flag. Override it here by looking into the job.
+	if r, ok := result.(*OSBuildJobResult); ok {
+		if !r.Success && r.OSBuildOutput != nil {
+			r.Success = r.OSBuildOutput.Success && len(r.TargetErrors) == 0
 		}
-	} else if !started.IsZero() {
-		state = common.CRunning
 	}
 
 	return &JobStatus{
-		State:    state,
 		Queued:   queued,
 		Started:  started,
 		Finished: finished,
 		Canceled: canceled,
-		Result:   result,
 	}, nil
 }
 
@@ -135,7 +116,11 @@ func (s *Server) Cancel(id uuid.UUID) error {
 // Provides access to artifacts of a job. Returns an io.Reader for the artifact
 // and the artifact's size.
 func (s *Server) JobArtifact(id uuid.UUID, name string) (io.Reader, int64, error) {
-	status, err := s.JobStatus(id)
+	if s.artifactsDir == "" {
+		return nil, 0, errors.New("Artifacts not enabled")
+	}
+
+	status, err := s.JobStatus(id, &json.RawMessage{})
 	if err != nil {
 		return nil, 0, err
 	}
@@ -160,7 +145,11 @@ func (s *Server) JobArtifact(id uuid.UUID, name string) (io.Reader, int64, error
 
 // Deletes all artifacts for job `id`.
 func (s *Server) DeleteArtifacts(id uuid.UUID) error {
-	status, err := s.JobStatus(id)
+	if s.artifactsDir == "" {
+		return errors.New("Artifacts not enabled")
+	}
+
+	status, err := s.JobStatus(id, &json.RawMessage{})
 	if err != nil {
 		return err
 	}
@@ -172,22 +161,29 @@ func (s *Server) DeleteArtifacts(id uuid.UUID) error {
 	return os.RemoveAll(path.Join(s.artifactsDir, id.String()))
 }
 
-func (s *Server) RequestOSBuildJob(ctx context.Context, arch string) (uuid.UUID, uuid.UUID, *OSBuildJob, error) {
+func (s *Server) RequestJob(ctx context.Context, arch string, jobTypes []string) (uuid.UUID, uuid.UUID, string, json.RawMessage, error) {
 	token := uuid.New()
 
-	// wait on "osbuild" jobs for backwards compatiblity
-	jobTypes := []string{"osbuild", "osbuild:" + arch}
+	// treat osbuild jobs specially until we have found a generic way to
+	// specify dequeuing restrictions. For now, we only have one
+	// restriction: arch for osbuild jobs.
+	jts := []string{}
+	for _, t := range jobTypes {
+		if t == "osbuild" {
+			t = "osbuild:" + arch
+		}
+		jts = append(jts, t)
+	}
 
-	var args OSBuildJob
-	jobId, err := s.jobs.Dequeue(ctx, jobTypes, &args)
+	jobId, jobType, args, err := s.jobs.Dequeue(ctx, jts)
 	if err != nil {
-		return uuid.Nil, uuid.Nil, nil, err
+		return uuid.Nil, uuid.Nil, "", nil, err
 	}
 
 	if s.artifactsDir != "" {
 		err := os.MkdirAll(path.Join(s.artifactsDir, "tmp", token.String()), 0700)
 		if err != nil {
-			return uuid.Nil, uuid.Nil, nil, fmt.Errorf("cannot create artifact directory: %v", err)
+			return uuid.Nil, uuid.Nil, "", nil, fmt.Errorf("cannot create artifact directory: %v", err)
 		}
 	}
 
@@ -195,7 +191,11 @@ func (s *Server) RequestOSBuildJob(ctx context.Context, arch string) (uuid.UUID,
 	defer s.runningMutex.Unlock()
 	s.running[token] = jobId
 
-	return token, jobId, &args, nil
+	if jobType == "osbuild:"+arch {
+		jobType = "osbuild"
+	}
+
+	return token, jobId, jobType, args, nil
 }
 
 func (s *Server) RunningJob(token uuid.UUID) (uuid.UUID, error) {
@@ -210,7 +210,7 @@ func (s *Server) RunningJob(token uuid.UUID) (uuid.UUID, error) {
 	return jobId, nil
 }
 
-func (s *Server) FinishJob(token uuid.UUID, result *OSBuildJobResult) error {
+func (s *Server) FinishJob(token uuid.UUID, result json.RawMessage) error {
 	s.runningMutex.Lock()
 	defer s.runningMutex.Unlock()
 
@@ -261,16 +261,7 @@ func (h *apiHandlers) RequestJob(ctx echo.Context) error {
 		return err
 	}
 
-	if len(body.Types) != 1 || body.Types[0] != "osbuild" {
-		return echo.NewHTTPError(http.StatusBadRequest, "invalid job types")
-	}
-
-	token, jobId, jobArgs, err := h.server.RequestOSBuildJob(ctx.Request().Context(), body.Arch)
-	if err != nil {
-		return err
-	}
-
-	serializedArgs, err := json.Marshal(jobArgs)
+	token, jobId, jobType, jobArgs, err := h.server.RequestJob(ctx.Request().Context(), body.Arch, body.Types)
 	if err != nil {
 		return err
 	}
@@ -279,8 +270,8 @@ func (h *apiHandlers) RequestJob(ctx echo.Context) error {
 		Id:               jobId,
 		Location:         fmt.Sprintf("%s/jobs/%v", api.BasePath, token),
 		ArtifactLocation: fmt.Sprintf("%s/jobs/%v/artifacts/", api.BasePath, token),
-		Type:             "osbuild",
-		Args:             serializedArgs,
+		Type:             jobType,
+		Args:             jobArgs,
 	})
 }
 
@@ -304,7 +295,7 @@ func (h *apiHandlers) GetJob(ctx echo.Context, tokenstr string) error {
 		return ctx.JSON(http.StatusOK, getJobResponse{})
 	}
 
-	status, err := h.server.JobStatus(jobId)
+	status, err := h.server.JobStatus(jobId, &json.RawMessage{})
 	if err != nil {
 		return err
 	}
@@ -326,14 +317,7 @@ func (h *apiHandlers) UpdateJob(ctx echo.Context, idstr string) error {
 		return err
 	}
 
-	// The jobqueue doesn't support setting the status before a job is
-	// finished. This branch should never be hit, because the worker
-	// doesn't attempt this. Change the API to remove this awkwardness.
-	if body.Status != common.IBFinished && body.Status != common.IBFailed {
-		return echo.NewHTTPError(http.StatusBadRequest, "setting status of a job to waiting or running is not supported")
-	}
-
-	err = h.server.FinishJob(token, &OSBuildJobResult{OSBuildOutput: body.Result})
+	err = h.server.FinishJob(token, body.Result)
 	if err != nil {
 		switch err {
 		case ErrTokenNotExist:


Test patch:
diff --git a/internal/common/states_test.go b/internal/common/states_test.go
index 8d0344d97..667c037b3 100644
--- a/internal/common/states_test.go
+++ b/internal/common/states_test.go
@@ -2,28 +2,26 @@ package common
 
 import (
 	"encoding/json"
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestJSONConversions(t *testing.T) {
 	type TestJson struct {
 		Ibs ImageBuildState `json:"ibs"`
-		Cs  ComposeState    `json:"cs"`
 	}
 	typedCases := []TestJson{
 		{
 			Ibs: IBWaiting,
-			Cs:  CWaiting,
 		},
 		{
 			Ibs: IBRunning,
-			Cs:  CFailed,
 		},
 	}
 	strCases := []string{
-		`{"ibs": "WAITING", "cs": "WAITING"}`,
-		`{"ibs": "RUNNING", "cs": "FAILED"}`,
+		`{"ibs": "WAITING"}`,
+		`{"ibs": "RUNNING"}`,
 	}
 
 	for n, c := range strCases {
diff --git a/internal/jobqueue/fsjobqueue/fsjobqueue_test.go b/internal/jobqueue/fsjobqueue/fsjobqueue_test.go
index 108da8677..4c3fec619 100644
--- a/internal/jobqueue/fsjobqueue/fsjobqueue_test.go
+++ b/internal/jobqueue/fsjobqueue/fsjobqueue_test.go
@@ -43,9 +43,11 @@ func pushTestJob(t *testing.T, q jobqueue.JobQueue, jobType string, args interfa
 }
 
 func finishNextTestJob(t *testing.T, q jobqueue.JobQueue, jobType string, result interface{}) uuid.UUID {
-	id, err := q.Dequeue(context.Background(), []string{jobType}, &json.RawMessage{})
+	id, typ, args, err := q.Dequeue(context.Background(), []string{jobType})
 	require.NoError(t, err)
 	require.NotEmpty(t, id)
+	require.Equal(t, jobType, typ)
+	require.NotNil(t, args)
 
 	err = q.FinishJob(id, result)
 	require.NoError(t, err)
@@ -89,16 +91,23 @@ func TestArgs(t *testing.T) {
 	twoargs := argument{42, "🐙"}
 	two := pushTestJob(t, q, "octopus", twoargs, nil)
 
-	var args argument
-	id, err := q.Dequeue(context.Background(), []string{"octopus"}, &args)
+	var parsedArgs argument
+
+	id, typ, args, err := q.Dequeue(context.Background(), []string{"octopus"})
 	require.NoError(t, err)
 	require.Equal(t, two, id)
-	require.Equal(t, twoargs, args)
+	require.Equal(t, "octopus", typ)
+	err = json.Unmarshal(args, &parsedArgs)
+	require.NoError(t, err)
+	require.Equal(t, twoargs, parsedArgs)
 
-	id, err = q.Dequeue(context.Background(), []string{"fish"}, &args)
+	id, typ, args, err = q.Dequeue(context.Background(), []string{"fish"})
 	require.NoError(t, err)
 	require.Equal(t, one, id)
-	require.Equal(t, oneargs, args)
+	require.Equal(t, "fish", typ)
+	err = json.Unmarshal(args, &parsedArgs)
+	require.NoError(t, err)
+	require.Equal(t, oneargs, parsedArgs)
 }
 
 func TestJobTypes(t *testing.T) {
@@ -113,9 +122,11 @@ func TestJobTypes(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
-	id, err := q.Dequeue(ctx, []string{"zebra"}, nil)
+	id, typ, args, err := q.Dequeue(ctx, []string{"zebra"})
 	require.Equal(t, err, context.Canceled)
 	require.Equal(t, uuid.Nil, id)
+	require.Equal(t, "", typ)
+	require.Nil(t, args)
 }
 
 func TestDependencies(t *testing.T) {
@@ -188,9 +199,11 @@ func TestMultipleWorkers(t *testing.T) {
 		defer close(done)
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer cancel()
-		id, err := q.Dequeue(ctx, []string{"octopus"}, &json.RawMessage{})
+		id, typ, args, err := q.Dequeue(ctx, []string{"octopus"})
 		require.NoError(t, err)
 		require.NotEmpty(t, id)
+		require.Equal(t, "octopus", typ)
+		require.Equal(t, json.RawMessage("null"), args)
 	}()
 
 	// Increase the likelihood that the above goroutine was scheduled and
@@ -199,9 +212,11 @@ func TestMultipleWorkers(t *testing.T) {
 
 	// 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{})
+	r, typ, args, err := q.Dequeue(context.Background(), []string{"clownfish"})
 	require.NoError(t, err)
 	require.Equal(t, id, r)
+	require.Equal(t, "clownfish", typ)
+	require.Equal(t, json.RawMessage("null"), args)
 
 	// Now wake up the Dequeue() in the goroutine and wait for it to finish.
 	_ = pushTestJob(t, q, "octopus", nil, nil)
@@ -230,9 +245,11 @@ func TestCancel(t *testing.T) {
 	// 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{})
+	r, typ, args, err := q.Dequeue(context.Background(), []string{"clownfish"})
 	require.NoError(t, err)
 	require.Equal(t, id, r)
+	require.Equal(t, "clownfish", typ)
+	require.Equal(t, json.RawMessage("null"), args)
 	err = q.CancelJob(id)
 	require.NoError(t, err)
 	_, _, _, canceled, err = q.JobStatus(id, &testResult{})
@@ -244,9 +261,11 @@ func TestCancel(t *testing.T) {
 	// 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{})
+	r, typ, args, err = q.Dequeue(context.Background(), []string{"clownfish"})
 	require.NoError(t, err)
 	require.Equal(t, id, r)
+	require.Equal(t, "clownfish", typ)
+	require.Equal(t, json.RawMessage("null"), args)
 	err = q.FinishJob(id, &testResult{})
 	require.NoError(t, err)
 	err = q.CancelJob(id)
diff --git a/internal/jobqueue/testjobqueue/testjobqueue.go b/internal/jobqueue/testjobqueue/testjobqueue.go
index 2c0d5e3f1..dc2635141 100644
--- a/internal/jobqueue/testjobqueue/testjobqueue.go
+++ b/internal/jobqueue/testjobqueue/testjobqueue.go
@@ -79,7 +79,7 @@ func (q *testJobQueue) Enqueue(jobType string, args interface{}, dependencies []
 	return j.Id, nil
 }
 
-func (q *testJobQueue) Dequeue(ctx context.Context, jobTypes []string, args interface{}) (uuid.UUID, error) {
+func (q *testJobQueue) Dequeue(ctx context.Context, jobTypes []string) (uuid.UUID, string, json.RawMessage, error) {
 	for _, t := range jobTypes {
 		if len(q.pending[t]) == 0 {
 			continue
@@ -90,16 +90,11 @@ func (q *testJobQueue) Dequeue(ctx context.Context, jobTypes []string, args inte
 
 		j := q.jobs[id]
 
-		err := json.Unmarshal(j.Args, args)
-		if err != nil {
-			return uuid.Nil, err
-		}
-
 		j.StartedAt = time.Now()
-		return j.Id, nil
+		return j.Id, j.Type, j.Args, nil
 	}
 
-	return uuid.Nil, errors.New("no job available")
+	return uuid.Nil, "", nil, errors.New("no job available")
 }
 
 func (q *testJobQueue) FinishJob(id uuid.UUID, result interface{}) error {
diff --git a/internal/weldr/api_test.go b/internal/weldr/api_test.go
index 31d3ed8dd..2057a1b50 100644
--- a/internal/weldr/api_test.go
+++ b/internal/weldr/api_test.go
@@ -474,15 +474,6 @@ func TestCompose(t *testing.T) {
 			QueueStatus: common.IBWaiting,
 			ImageType:   imgType,
 			Manifest:    manifest,
-			Targets: []*target.Target{
-				{
-					// skip Uuid and Created fields - they are ignored
-					Name: "org.osbuild.local",
-					Options: &target.LocalTargetOptions{
-						Filename: "test.img",
-					},
-				},
-			},
 		},
 	}
 	expectedComposeLocalAndAws := &store.Compose{
@@ -512,13 +503,6 @@ func TestCompose(t *testing.T) {
 						Key:             "imagekey",
 					},
 				},
-				{
-					// skip Uuid and Created fields - they are ignored
-					Name: "org.osbuild.local",
-					Options: &target.LocalTargetOptions{
-						Filename: "test.img",
-					},
-				},
 			},
 		},
 	}
diff --git a/internal/worker/server_test.go b/internal/worker/server_test.go
index c89836e53..56aa29668 100644
--- a/internal/worker/server_test.go
+++ b/internal/worker/server_test.go
@@ -64,7 +64,7 @@ func TestCreate(t *testing.T) {
 	}
 	server := worker.NewServer(nil, testjobqueue.New(), "")
 
-	_, err = server.Enqueue(arch.Name(), manifest, nil)
+	_, err = server.Enqueue(arch.Name(), &worker.OSBuildJob{Manifest: manifest})
 	require.NoError(t, err)
 
 	test.TestRoute(t, server, false, "POST", "/api/worker/v1/jobs", `{"types":["osbuild"],"arch":"x86_64"}`, http.StatusCreated,
@@ -87,12 +87,14 @@ func TestCancel(t *testing.T) {
 	}
 	server := worker.NewServer(nil, testjobqueue.New(), "")
 
-	jobId, err := server.Enqueue(arch.Name(), manifest, nil)
+	jobId, err := server.Enqueue(arch.Name(), &worker.OSBuildJob{Manifest: manifest})
 	require.NoError(t, err)
 
-	token, j, _, err := server.RequestOSBuildJob(context.Background(), arch.Name())
+	token, j, typ, args, err := server.RequestJob(context.Background(), arch.Name(), []string{"osbuild"})
 	require.NoError(t, err)
 	require.Equal(t, jobId, j)
+	require.Equal(t, "osbuild", typ)
+	require.NotNil(t, args)
 
 	err = server.Cancel(jobId)
 	require.NoError(t, err)
diff --git a/test/cases/api.sh b/test/cases/api.sh
index 28c8971a7..d4ac9371f 100644
--- a/test/cases/api.sh
+++ b/test/cases/api.sh
@@ -143,8 +143,8 @@ do
 
   COMPOSE_STATUS=$(echo "$OUTPUT" | jq -r '.status')
 
-  if [[ "$COMPOSE_STATUS" != "WAITING" && "$COMPOSE_STATUS" != "RUNNING" ]]; then
-    test "$COMPOSE_STATUS" = "FINISHED"
+  if [[ "$COMPOSE_STATUS" != "pending" && "$COMPOSE_STATUS" != "running" ]]; then
+    test "$COMPOSE_STATUS" = "success"
     break
   fi
 
diff --git a/test/cases/koji.sh b/test/cases/koji.sh
index 60b1527ed..79b8ff66b 100755
--- a/test/cases/koji.sh
+++ b/test/cases/koji.sh
@@ -14,13 +14,6 @@ function greenprint {
 # Provision the software under tet.
 /usr/libexec/osbuild-composer-test/provision.sh
 
-if [[ $ID == rhel ]] && ! rpm -q epel-release; then
-    greenprint "📦 Setting up EPEL repository"
-    curl -Ls --retry 5 --output /tmp/epel.rpm \
-        https://dl.fedoraproject.org/pub/epel/epel-release-latest-8.noarch.rpm
-    sudo rpm -Uvh /tmp/epel.rpm
-fi
-
 greenprint "Adding podman dnsname plugin"
 if [[ $ID == rhel ]]; then
   sudo cp /usr/share/tests/osbuild-composer/vendor/87-podman-bridge.conflist /etc/cni/net.d/
diff --git a/test/cases/ostree.sh b/test/cases/ostree.sh
index 79c250970..702c92f25 100755
--- a/test/cases/ostree.sh
+++ b/test/cases/ostree.sh
@@ -33,14 +33,6 @@ function greenprint {
     echo -e "\033[1;32m${1}\033[0m"
 }
 
-# Mock is only available in EPEL for RHEL.
-if [[ $ID == rhel ]] && ! rpm -q epel-release; then
-    greenprint "📦 Setting up EPEL repository"
-    curl -Ls --retry 5 --output /tmp/epel.rpm \
-        https://dl.fedoraproject.org/pub/epel/epel-release-latest-8.noarch.rpm
-    sudo rpm -Uvh /tmp/epel.rpm
-fi
-
 # Start libvirtd and test it.
 greenprint "🚀 Starting libvirt daemon"
 sudo systemctl start libvirtd