Repository: osbuild/osbuild-composer
Pull request #3766 (2023-10-27 13:30:14)

Worker status heartbeat
    Unresponsive workers (>=1 hour of no status update) are cleaned up.
    
    Several things are enabled by keeping track of workers, in future the
    worker server could:
    - keep track of how many workers are active
    - see if a worker for a specific architecture is available



henrywang: @croissanne Test got error `Nov 15 13:06:15 runner-rhos-0x-rhel-8-9-large-7619 osbuild-worker[23989]: time="2023-11-15T13:06:15-05:00" level=error msg="Requesting job failed: error requesting job: 500 — Error is not an instance of HTTPError (IMAGE-BUILDER-WORKER-10001)"` when build ostree upgrade image. 
croissanne: > @croissanne Test got error `Nov 15 13:06:15 runner-rhos-0x-rhel-8-9-large-7619 osbuild-worker[23989]: time="2023-11-15T13:06:15-05:00" level=error msg="Requesting job failed: error requesting job: 500 — Error is not an instance of HTTPError (IMAGE-BUILDER-WORKER-10001)"` when build ostree upgrade image.

Hm is the full worker log in the workflow?
henrywang: @croissanne All Edge tests have been moved to RHEL 9.4 and 8.10 with PR https://github.com/osbuild/osbuild-composer/pull/3808. Could you please a rebase to make test green? Thanks.

Patch:
diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index 8c01a4f86..86d2d86db 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -98,6 +98,8 @@ RPM:
           - aws/fedora-37-aarch64
           - aws/fedora-38-x86_64
           - aws/fedora-38-aarch64
+          - aws/fedora-40-x86_64
+          - aws/fedora-40-aarch64
           - aws/rhel-8.4-ga-x86_64
           - aws/rhel-8.4-ga-aarch64
           - aws/rhel-8.8-ga-x86_64
diff --git a/Schutzfile b/Schutzfile
index 45db501e3..c521d2604 100644
--- a/Schutzfile
+++ b/Schutzfile
@@ -153,6 +153,32 @@
       }
     ]
   },
+  "fedora-40": {
+    "dependencies": {
+      "osbuild": {
+        "commit": "36883654cea32446ba60747a47d23ef38ca200f4"
+      }
+    },
+    "repos": [
+      {
+        "file": "/etc/yum.repos.d/fedora.repo",
+        "x86_64": [
+          {
+            "title": "fedora",
+            "name": "fedora",
+            "baseurl": "https://rpmrepo.osbuild.org/v2/mirror/public/f40/f40-x86_64-rawhide-20231201"
+          }
+        ],
+        "aarch64": [
+          {
+            "title": "fedora",
+            "name": "fedora",
+            "baseurl": "https://rpmrepo.osbuild.org/v2/mirror/public/f40/f40-aarch64-rawhide-20231201"
+          }
+        ]
+      }
+    ]
+  },
   "rhel-8.4": {
     "dependencies": {
       "osbuild": {
diff --git a/internal/jobqueue/fsjobqueue/fsjobqueue.go b/internal/jobqueue/fsjobqueue/fsjobqueue.go
index 57252cb61..69b7ca8ea 100644
--- a/internal/jobqueue/fsjobqueue/fsjobqueue.go
+++ b/internal/jobqueue/fsjobqueue/fsjobqueue.go
@@ -56,6 +56,15 @@ type fsJobQueue struct {
 	// reported as done.
 	jobIdByToken map[uuid.UUID]uuid.UUID
 	heartbeats   map[uuid.UUID]time.Time // token -> heartbeat
+
+	workerIDByToken map[uuid.UUID]uuid.UUID // token -> workerID
+	workers         map[uuid.UUID]worker
+}
+
+type worker struct {
+	Arch      string    `json:"arch"`
+	Heartbeat time.Time `json:"heartbeat"`
+	Tokens    map[uuid.UUID]struct{}
 }
 
 // On-disk job struct. Contains all necessary (but non-redundant) information
@@ -85,12 +94,14 @@ type job struct {
 // loaded and rescheduled to run if necessary.
 func New(dir string) (*fsJobQueue, error) {
 	q := &fsJobQueue{
-		db:           jsondb.New(dir, 0600),
-		pending:      list.New(),
-		dependants:   make(map[uuid.UUID][]uuid.UUID),
-		jobIdByToken: make(map[uuid.UUID]uuid.UUID),
-		heartbeats:   make(map[uuid.UUID]time.Time),
-		listeners:    make(map[chan struct{}]struct{}),
+		db:              jsondb.New(dir, 0600),
+		pending:         list.New(),
+		dependants:      make(map[uuid.UUID][]uuid.UUID),
+		jobIdByToken:    make(map[uuid.UUID]uuid.UUID),
+		heartbeats:      make(map[uuid.UUID]time.Time),
+		listeners:       make(map[chan struct{}]struct{}),
+		workers:         make(map[uuid.UUID]worker),
+		workerIDByToken: make(map[uuid.UUID]uuid.UUID),
 	}
 
 	// Look for jobs that are still pending and build the dependant map.
@@ -186,7 +197,7 @@ func (q *fsJobQueue) Enqueue(jobType string, args interface{}, dependencies []uu
 	return j.Id, nil
 }
 
-func (q *fsJobQueue) Dequeue(ctx context.Context, jobTypes []string, channels []string) (uuid.UUID, uuid.UUID, []uuid.UUID, string, json.RawMessage, error) {
+func (q *fsJobQueue) Dequeue(ctx context.Context, wID uuid.UUID, jobTypes, channels []string) (uuid.UUID, uuid.UUID, []uuid.UUID, string, json.RawMessage, error) {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
@@ -231,6 +242,10 @@ func (q *fsJobQueue) Dequeue(ctx context.Context, jobTypes []string, channels []
 	j.Token = uuid.New()
 	q.jobIdByToken[j.Token] = j.Id
 	q.heartbeats[j.Token] = time.Now()
+	if _, ok := q.workers[wID]; ok {
+		q.workers[wID].Tokens[j.Token] = struct{}{}
+		q.workerIDByToken[j.Token] = wID
+	}
 
 	err := q.db.Write(j.Id.String(), j)
 	if err != nil {
@@ -240,7 +255,7 @@ func (q *fsJobQueue) Dequeue(ctx context.Context, jobTypes []string, channels []
 	return j.Id, j.Token, j.Dependencies, j.Type, j.Args, nil
 }
 
-func (q *fsJobQueue) DequeueByID(ctx context.Context, id uuid.UUID) (uuid.UUID, []uuid.UUID, string, json.RawMessage, error) {
+func (q *fsJobQueue) DequeueByID(ctx context.Context, id, wID uuid.UUID) (uuid.UUID, []uuid.UUID, string, json.RawMessage, error) {
 	q.mu.Lock()
 	defer q.mu.Unlock()
 
@@ -268,6 +283,10 @@ func (q *fsJobQueue) DequeueByID(ctx context.Context, id uuid.UUID) (uuid.UUID,
 	j.Token = uuid.New()
 	q.jobIdByToken[j.Token] = j.Id
 	q.heartbeats[j.Token] = time.Now()
+	if _, ok := q.workers[wID]; ok {
+		q.workers[wID].Tokens[j.Token] = struct{}{}
+		q.workerIDByToken[j.Token] = wID
+	}
 
 	err = q.db.Write(j.Id.String(), j)
 	if err != nil {
@@ -296,6 +315,10 @@ func (q *fsJobQueue) RequeueOrFinishJob(id uuid.UUID, maxRetries uint64, result
 
 	delete(q.jobIdByToken, j.Token)
 	delete(q.heartbeats, j.Token)
+	if wID, ok := q.workerIDByToken[j.Token]; ok {
+		delete(q.workers[wID].Tokens, j.Token)
+		delete(q.workerIDByToken, j.Token)
+	}
 
 	if j.Retries >= maxRetries {
 		j.FinishedAt = time.Now()
@@ -444,6 +467,62 @@ func (q *fsJobQueue) RefreshHeartbeat(token uuid.UUID) {
 	}
 }
 
+func (q *fsJobQueue) InsertWorker(arch string) (uuid.UUID, error) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	wID := uuid.New()
+	q.workers[wID] = worker{
+		Arch:      arch,
+		Heartbeat: time.Now(),
+		Tokens:    make(map[uuid.UUID]struct{}),
+	}
+	return wID, nil
+}
+
+func (q *fsJobQueue) UpdateWorkerStatus(wID uuid.UUID) error {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	worker, ok := q.workers[wID]
+	if !ok {
+		return jobqueue.ErrWorkerNotExist
+	}
+
+	worker.Heartbeat = time.Now()
+	return nil
+}
+
+func (q *fsJobQueue) Workers(olderThan time.Duration) ([]uuid.UUID, error) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	now := time.Now()
+	wIDs := []uuid.UUID{}
+	for wID, worker := range q.workers {
+		if now.Sub(worker.Heartbeat) > olderThan {
+			wIDs = append(wIDs, wID)
+		}
+	}
+	return wIDs, nil
+}
+
+func (q *fsJobQueue) DeleteWorker(wID uuid.UUID) error {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+
+	worker, ok := q.workers[wID]
+	if !ok {
+		return jobqueue.ErrWorkerNotExist
+	}
+
+	if len(worker.Tokens) != 0 {
+		return jobqueue.ErrActiveJobs
+	}
+	delete(q.workers, wID)
+	return 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) {
diff --git a/internal/worker/api/api.gen.go b/internal/worker/api/api.gen.go
index 32bcbae9f..12241ac37 100644
--- a/internal/worker/api/api.gen.go
+++ b/internal/worker/api/api.gen.go
@@ -51,10 +51,24 @@ type ObjectReference struct {
 	Kind string `json:"kind"`
 }
 
+// PostWorkersRequest defines model for PostWorkersRequest.
+type PostWorkersRequest struct {
+	Arch string `json:"arch"`
+}
+
+// PostWorkersResponse defines model for PostWorkersResponse.
+type PostWorkersResponse struct {
+	// Embedded struct due to allOf(#/components/schemas/ObjectReference)
+	ObjectReference `yaml:",inline"`
+	// Embedded fields due to inline allOf schema
+	WorkerId string `json:"worker_id"`
+}
+
 // RequestJobRequest defines model for RequestJobRequest.
 type RequestJobRequest struct {
-	Arch  string   `json:"arch"`
-	Types []string `json:"types"`
+	Arch     string   `json:"arch"`
+	Types    []string `json:"types"`
+	WorkerId *string  `json:"worker_id,omitempty"`
 }
 
 // RequestJobResponse defines model for RequestJobResponse.
@@ -91,12 +105,18 @@ type RequestJobJSONBody RequestJobRequest
 // UpdateJobJSONBody defines parameters for UpdateJob.
 type UpdateJobJSONBody UpdateJobRequest
 
+// PostWorkersJSONBody defines parameters for PostWorkers.
+type PostWorkersJSONBody PostWorkersRequest
+
 // RequestJobJSONRequestBody defines body for RequestJob for application/json ContentType.
 type RequestJobJSONRequestBody RequestJobJSONBody
 
 // UpdateJobJSONRequestBody defines body for UpdateJob for application/json ContentType.
 type UpdateJobJSONRequestBody UpdateJobJSONBody
 
+// PostWorkersJSONRequestBody defines body for PostWorkers for application/json ContentType.
+type PostWorkersJSONRequestBody PostWorkersJSONBody
+
 // ServerInterface represents all server handlers.
 type ServerInterface interface {
 	// Get error description
@@ -120,6 +140,12 @@ type ServerInterface interface {
 	// status
 	// (GET /status)
 	GetStatus(ctx echo.Context) error
+	// Create a new worker
+	// (POST /workers)
+	PostWorkers(ctx echo.Context) error
+	// Refresh worker status
+	// (POST /workers/{worker_id}/status)
+	PostWorkerStatus(ctx echo.Context, workerId string) error
 }
 
 // ServerInterfaceWrapper converts echo contexts to parameters.
@@ -228,6 +254,31 @@ func (w *ServerInterfaceWrapper) GetStatus(ctx echo.Context) error {
 	return err
 }
 
+// PostWorkers converts echo context to params.
+func (w *ServerInterfaceWrapper) PostWorkers(ctx echo.Context) error {
+	var err error
+
+	// Invoke the callback with all the unmarshalled arguments
+	err = w.Handler.PostWorkers(ctx)
+	return err
+}
+
+// PostWorkerStatus converts echo context to params.
+func (w *ServerInterfaceWrapper) PostWorkerStatus(ctx echo.Context) error {
+	var err error
+	// ------------- Path parameter "worker_id" -------------
+	var workerId string
+
+	err = runtime.BindStyledParameterWithLocation("simple", false, "worker_id", runtime.ParamLocationPath, ctx.Param("worker_id"), &workerId)
+	if err != nil {
+		return echo.NewHTTPError(http.StatusBadRequest, fmt.Sprintf("Invalid format for parameter worker_id: %s", err))
+	}
+
+	// Invoke the callback with all the unmarshalled arguments
+	err = w.Handler.PostWorkerStatus(ctx, workerId)
+	return err
+}
+
 // This is a simple interface which specifies echo.Route addition functions which
 // are present on both echo.Echo and echo.Group, since we want to allow using
 // either of them for path registration
@@ -263,32 +314,39 @@ func RegisterHandlersWithBaseURL(router EchoRouter, si ServerInterface, baseURL
 	router.PUT(baseURL+"/jobs/:token/artifacts/:name", wrapper.UploadJobArtifact)
 	router.GET(baseURL+"/openapi", wrapper.GetOpenapi)
 	router.GET(baseURL+"/status", wrapper.GetStatus)
+	router.POST(baseURL+"/workers", wrapper.PostWorkers)
+	router.POST(baseURL+"/workers/:worker_id/status", wrapper.PostWorkerStatus)
 
 }
 
 // Base64 encoded, gzipped, json marshaled Swagger object
 var swaggerSpec = []string{
 
-	"H4sIAAAAAAAC/9xYX2/bNhD/KgS3hw2QLadpXwTsoemGIh26DMmKFciC4EydLSYSqZAnO4ah7z6QlP9J",
-	"ip0A8UPzJFk83p/f/Xh39JILXZRaoSLLkyW3IsMC/OsfxmjjXiDPLyY8uV7ynw1OeMJ/ijeb4mZHfDG+",
-	"Q0GXOEGDSiCvoyUvjS7RkESvUOgU3ZMWJfKEWzJSTXkd8QKthalfS9EKI0uSWvGEn4G4n4NJmbMHJMcy",
-	"l7Rgc0kZm2tzj8ay/6rR6FT8xmanpxHDhwpyywyC1YpHXVPOH3Dab2Xa60uztbvk1x4qaTDlyXUIZi3e",
-	"UrwJ6Wbtg/b48PqmjvhnpC96fIm21Mriq2IMSmCO27GNtc4RVDeClWi/j21bSdtU5h3tgfAJZO+lSg/j",
-	"6tHzolGw0PUu4pf4UKENGPq3rndgRNbrhvvgJSRhYZ8U4QkHY2DRcTDsj4KBQ869foLBTP3zcTDVg8b2",
-	"ndVqeAnzrw3paucdyQkIus21gHCaegJNFwoKKW5XSteQHNC+C1DE9xoJHw7l3a9uaeoLoZ+oVwRU2WNg",
-	"bb3mw743cv3ufStTINxHVYO2yukg7C2jza4+Bm6Z3IDyIiicMakmuluS/8mkZdIyUOzj3+dsos26EpNm",
-	"JsTIQKUsA5XmyO702A5dKZaUOzcvrs4qmafsk3PDomED9q9XwCM+Q2ODmZOmWCsoJU/46XA0HPGIl0CZ",
-	"xyxG151svJRp7X5Pkbq+fkbnCZPKkqt1TE8YZcj8VmZLFHIiMWXjBfNVZ13Cz9OwOXRAZ9VAgYTGelLt",
-	"Gjn/fUcvd8DxxHvKI66gcEF7/ZvskakwanqtcxsfoSg9Oien3a5V37i9IZM++HejUeinilD5uKEscxlO",
-	"SXzX9K+N+n2pDzHWPuPvv38/it4PR9FbR9yiqIykhU/LGYJBw5PrGweYrYoCzKJhQUj5duLc9thx059H",
-	"bXvo0xxYy8CReMg89dckYeNci3vLKkUyDyL+XMxA5jDOcdhh1KYxNGRAS2c6XbwaNt22GGBqkefkKAab",
-	"SuMN7uL4ySAQpu5Evxu9fzXjvUVr1/Jf2qdlDlt5iRiZBYMpSMV/NM634/Ms3jD9clV9XdQbhsdL0veo",
-	"tutkp9StSHmkKtMaeHtCufiT/5AVaKfMmEopqaYB/k7f6OkLPjF7W0NPLyiBwmy7m8V11z9SdekMMr3F",
-	"ZXQMe2+YNiFKBrvcaR/deDUM23jpqOPPcllRHwtyDekXPf7Y7ODP4aF/vISG0evR+Xlc1YKQBpYMQrEL",
-	"elvlU6R8c8RxiXbz7YobgTbrofnpYn/RiDwHp0adH5eZVMz57qb+AvxV48MxRtH2If+m8LFEQZg2g5wW",
-	"ojKOX90S7AbxvT47jDYXu957w5V00zgLUs09xrB5JkXGDFJllGUWzUyKlVDf7eFqtXK0Ctm6+b7F8tjA",
-	"20z7ZtZ/B/sKUrFfSqPTSrhPv7IgyyNemZwnPCMqbRLHUMqhY4fN5ISGQhfuSywLmOJg7K6laAbhOhvP",
-	"Tvw/Ai1mEExdkd6j3hJM8YVGgpaXiG0t3NT/BwAA//+IToO0xRUAAA==",
+	"H4sIAAAAAAAC/9xY224bvRF+lQFboC2w1spxeiOgF3FaBE6ROrAb/AFiI5jdHWlpc8kNOWtFEPTuP0iu",
+	"jruSbMC6SK5kS8M5fPNxDpyL3FS10aTZidFcuLykCsOf/7HWWP8HKnU9FqNvc/FXS2MxEn9J14fS9kR6",
+	"nT1Qzjc0Jks6J7FI5qK2pibLkoLC3BTkP3lWkxgJx1bqiVgkoiLncBJ+K8jlVtYsjRYjcYn54xRtAd4e",
+	"ssykkjyDqeQSpsY+knVw1wyHF/m/4OniIgH60aByYAmd0SLpmvL+oNf+XRa9vrRHuz+F33400lIhRt9i",
+	"MCvxHcXrkO5XPpiAj1jcLxLxgfijyW7I1UY7elWMUeekaDO2zBhFqLsRLEX7fdy1Ndo1VQZHeyDcg+yj",
+	"1MVxXAN6QTSJFrreJeKzcfxHzP8N/WjIcdc9tHl53FyQOmri9fMU2duScGxshSxGomlC+IddXh/tT1yL",
+	"SCDYy7CJ6oKEZKrcXhExEmgtzvz/W5Ec9jxqT/aDvun662OOdhI+f55NzFlr+8EZPbjB6af2vi68dyzH",
+	"mPN3ZXKMhagHhmKmsZL596XSFWBHtO/Cd9BI/OI5sIoNTX0h9FPllpGbk/DbBc3HfW/l+t37UhfIdIjI",
+	"llyj+CjsO0bbU30M3DC5BuVFUHhjUo9Nt5v9v5QOpAPU8O7zFYyNXTUxNmBjjIC6gBJ1oQgeTOYGvh5I",
+	"Vt7N69vLRqoC3ns3HFk4g1iiRCKeyLpo5rztcxprKUbiYjAcDEUiauQyYJaSb+wuncti4f+fEHd9/UDe",
+	"E5DasW8TYMbAJUE4Cq6mXI4lFZDNIFSsVfe7KuLhODx4qxYrYrIukGrbyNW/t/QKD5wYBU9FIjRWPuig",
+	"f509tg0l7Zji3aafWNUBnfOLntJ578/GTIbg3wyHcRTRTDrEjXWtZLwl6UPb+tfqD6U+xrgIGX/79etJ",
+	"9P7zJHoXiXCUN1byLKTlktCSFaNv9x4w11QV2lnLgpjyzcT546nnZriPxvXQp72wDtCTeACB+iuSQKZM",
+	"/uig0SxVFAn34gmlwkzRoMOodWNoyUCOL00xezVsuk0zwrRDnvOTGGwrTTC4jeN7S8hU+Bv9Zvj21Yz3",
+	"Fq1ty/8zIS1T3MhLAmxngBOUWvxqnN+NL7B4zfSbZfX1Ua8Zns7ZPJLerJOdUrck5YmqzM6u0BPK9X/F",
+	"L1mBtsqMbbSWehLh7/SNnr4QEnOwNfT0gho5Tr7bWVx1/RNVl84g01tchqew9xvTJkYJuM2d3aubLodh",
+	"l849dcJdrhvuY4EyWHw02bv2hHgOD8PHS2iYvB6dn8dVkzPxmWNLWG2DvqtyHyl/O+L4RPv5dsmNSJvV",
+	"0Ly/2F+3Is/BqVUXxmWQGrzv0O76PtBTjKK7l/yLpp815UxFO8iZPG+s51e3BPtB/KDPHqP1Yte7N9xK",
+	"P41DlGr3GAvTUuYlWOLGageO7JPMl0J928Pt8peTVcidzfd3LI8tvCFr7Z65f2CPg6af1zVN27U0rKLL",
+	"pCE0jSzaTLrSNKqAjKBxVHieoFLgmsz5gqQZclTKDe50J7kbz2on6rU9b4MnHuX7ngr3z/JbEO/cwijS",
+	"lVjmL52vXtsWGzfxeJNaP9Id6jDH3iH9ALVn3RtbciW5UENKQssZIS+3+2g9CXzyCly4+l4YPVX0ikhs",
+	"oPBxVFITmCeyqNSdbtlYEiouB+2TR3u6MKANQ8xtAVOpVPgiI3ikmoEt5o/eDxyzZzSwrMg0PICr8Z0u",
+	"rKlrKlbPMFOytJomwiqQAPvVNaxCXndG3hajZSoO8/tgDduGzzV5Tm7cKDWDJgw1S5f+5mDjGm+uKwHw",
+	"5UVdyYTF3j71P7d8Qqnh77U1RZP7r/4BUVYkorFKjETJXLtRmmItB74RuFKOeZCbyn+TygondJY1UhVk",
+	"z6Ll9Ok8PP7tNAHGiUfwgHrHOKEXGolaXiK28cP94s8AAAD//7D+9ubrGgAA",
 }
 
 // GetSwagger returns the content of the embedded swagger specification file
diff --git a/internal/worker/api/errors.go b/internal/worker/api/errors.go
index 55c9524d7..157dea08c 100644
--- a/internal/worker/api/errors.go
+++ b/internal/worker/api/errors.go
@@ -23,7 +23,8 @@ const (
 	ErrorErrorNotFound        ServiceErrorCode = 14
 	ErrorInvalidJobType       ServiceErrorCode = 15
 	ErrorTenantNotFound       ServiceErrorCode = 16
-	// ErrorTokenNotFound ServiceErrorCode = 6
+	ErrorMalformedWorkerId    ServiceErrorCode = 17
+	ErrorWorkerIdNotFound     ServiceErrorCode = 18
 
 	// internal errors
 	ErrorDiscardingArtifact       ServiceErrorCode = 1000
@@ -34,6 +35,8 @@ const (
 	ErrorRetrievingJobStatus      ServiceErrorCode = 1005
 	ErrorRequestingJob            ServiceErrorCode = 1006
 	ErrorFailedLoadingOpenAPISpec ServiceErrorCode = 1007
+	ErrorInsertingWorker          ServiceErrorCode = 1008
+	ErrorUpdatingWorkerStatus     ServiceErrorCode = 1009
 
 	// Errors contained within this file
 	ErrorUnspecified          ServiceErrorCode = 10000
@@ -56,11 +59,20 @@ func getServiceErrors() serviceErrors {
 	return serviceErrors{
 		serviceError{ErrorUnsupportedMediaType, http.StatusUnsupportedMediaType, "Only 'application/json' content is supported"},
 		serviceError{ErrorBodyDecodingError, http.StatusBadRequest, "Malformed json, unable to decode body"},
-
 		serviceError{ErrorJobNotFound, http.StatusNotFound, "Token not found"},
 		serviceError{ErrorJobNotRunning, http.StatusBadRequest, "Job is not running"},
 		serviceError{ErrorMalformedJobId, http.StatusBadRequest, "Given job id is not a uuidv4"},
 		serviceError{ErrorMalformedJobToken, http.StatusBadRequest, "Given job id is not a uuidv4"},
+		serviceError{ErrorInvalidErrorId, http.StatusBadRequest, "Invalid format for error id, it should be an integer as a string"},
+		serviceError{ErrorResourceNotFound, http.StatusNotFound, "Requested resource doesn't exist"},
+		serviceError{ErrorMethodNotAllowed, http.StatusMethodNotAllowed, "Requested method isn't supported for resource"},
+		serviceError{ErrorNotAcceptable, http.StatusNotAcceptable, "Only 'application/json' content is supported"},
+		serviceError{ErrorErrorNotFound, http.StatusNotFound, "Error with given id not found"},
+		serviceError{ErrorInvalidJobType, http.StatusBadRequest, "Requested job type cannot be dequeued"},
+		serviceError{ErrorTenantNotFound, http.StatusBadRequest, "Tenant not found in JWT claims"},
+		serviceError{ErrorMalformedWorkerId, http.StatusBadRequest, "Given worker id is not a uuidv4"},
+		serviceError{ErrorWorkerIdNotFound, http.StatusBadRequest, "Given worker id doesn't exist"},
+
 		serviceError{ErrorDiscardingArtifact, http.StatusInternalServerError, "Error discarding artifact"},
 		serviceError{ErrorCreatingArtifact, http.StatusInternalServerError, "Error creating artifact"},
 		serviceError{ErrorWritingArtifact, http.StatusInternalServerError, "Error writing artifact"},
@@ -68,14 +80,9 @@ func getServiceErrors() serviceErrors {
 		serviceError{ErrorFinishingJob, http.StatusInternalServerError, "Error finishing job"},
 		serviceError{ErrorRetrievingJobStatus, http.StatusInternalServerError, "Error requesting job"},
 		serviceError{ErrorRequestingJob, http.StatusInternalServerError, "Error requesting job"},
-		serviceError{ErrorInvalidErrorId, http.StatusBadRequest, "Invalid format for error id, it should be an integer as a string"},
 		serviceError{ErrorFailedLoadingOpenAPISpec, http.StatusInternalServerError, "Unable to load openapi spec"},
-		serviceError{ErrorResourceNotFound, http.StatusNotFound, "Requested resource doesn't exist"},
-		serviceError{ErrorMethodNotAllowed, http.StatusMethodNotAllowed, "Requested method isn't supported for resource"},
-		serviceError{ErrorNotAcceptable, http.StatusNotAcceptable, "Only 'application/json' content is supported"},
-		serviceError{ErrorErrorNotFound, http.StatusNotFound, "Error with given id not found"},
-		serviceError{ErrorInvalidJobType, http.StatusBadRequest, "Requested job type cannot be dequeued"},
-		serviceError{ErrorTenantNotFound, http.StatusBadRequest, "Tenant not found in JWT claims"},
+		serviceError{ErrorInsertingWorker, http.StatusInternalServerError, "Unable to register the worker"},
+		serviceError{ErrorUpdatingWorkerStatus, http.StatusInternalServerError, "Unable update worker status"},
 
 		serviceError{ErrorUnspecified, http.StatusInternalServerError, "Unspecified internal error "},
 		serviceError{ErrorNotHTTPError, http.StatusInternalServerError, "Error is not an instance of HTTPError"},
diff --git a/internal/worker/api/openapi.yml b/internal/worker/api/openapi.yml
index 7f19aedd8..cccca975a 100644
--- a/internal/worker/api/openapi.yml
+++ b/internal/worker/api/openapi.yml
@@ -1,3 +1,4 @@
+
 openapi: 3.0.0
 info:
   title: OSBuild Composer - Worker
@@ -203,6 +204,43 @@ paths:
               schema:
                 $ref: '#/components/schemas/Error'
 
+  /workers:
+    post:
+      operationId: postWorkers
+      summary: Create a new worker
+      description: |
+        Creates a new worker and returns a uuid which should be used in all subsequent calls.
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/PostWorkersRequest'
+      responses:
+        '201':
+          description: Created a new worker
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/PostWorkersResponse'
+  /workers/{worker_id}/status:
+    parameters:
+      - schema:
+          type: string
+          format: uuid
+        name: worker_id
+        in: path
+        required: true
+    post:
+      operationId: postWorkerStatus
+      summary: Refresh worker status
+      description: |
+        Refreshes the heartbeat of the worker, and posts stats that can be used to determine overall
+        worker health. Workers that do not respond will not be kept track of after a timeout. If
+        dropped workers were running a job, this job will be restarted.
+      responses:
+        '200':
+          description: succesfully updated worker's status
+
 components:
   schemas:
     ObjectReference:
@@ -261,6 +299,8 @@ components:
             type: string
         arch:
           type: string
+        worker_id:
+          type: string
     RequestJobResponse:
       allOf:
       - $ref: '#/components/schemas/ObjectReference'
@@ -301,3 +341,21 @@ components:
           x-go-type: json.RawMessage
     UpdateJobResponse:
       $ref: '#/components/schemas/ObjectReference'
+
+    PostWorkersRequest:
+      type: object
+      required:
+        - arch
+      properties:
+        arch:
+          type: string
+    PostWorkersResponse:
+      allOf:
+      - $ref: '#/components/schemas/ObjectReference'
+      - type: object
+        required:
+          - worker_id
+        properties:
+          worker_id:
+            type: string
+            format: uuid
diff --git a/internal/worker/client.go b/internal/worker/client.go
index d35cc6713..32bc93224 100644
--- a/internal/worker/client.go
+++ b/internal/worker/client.go
@@ -12,9 +12,12 @@ import (
 	"net/http"
 	"net/url"
 	"sync"
+	"time"
 
 	"github.com/google/uuid"
+	"github.com/sirupsen/logrus"
 
+	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/worker/api"
 )
 
@@ -26,8 +29,10 @@ type Client struct {
 	accessToken  string
 	clientId     string
 	clientSecret string
+	workerID     uuid.UUID
 
-	tokenMu sync.RWMutex
+	tokenMu    sync.RWMutex
+	workerIDMu sync.RWMutex
 }
 
 type ClientConfig struct {
@@ -49,7 +54,7 @@ type Job interface {
 	NDynamicArgs() int
 	Update(result interface{}) error
 	Canceled() (bool, error)
-	UploadArtifact(name string, reader io.Reader) error
+	UploadArtifact(name string, readSeeker io.ReadSeeker) error
 }
 
 var ErrClientRequestJobTimeout = errors.New("Dequeue timed out, retry")
@@ -108,14 +113,20 @@ func NewClient(conf ClientConfig) (*Client, error) {
 	}
 	requester.Transport = transport
 
-	return &Client{
+	client := &Client{
 		server:       server,
 		requester:    requester,
 		offlineToken: conf.OfflineToken,
 		oAuthURL:     conf.OAuthURL,
 		clientId:     conf.ClientId,
 		clientSecret: conf.ClientSecret,
-	}, nil
+	}
+	err = client.registerWorker()
+	if err != nil {
+		return client, err
+	}
+	go client.workerHeartbeat()
+	return client, nil
 }
 
 func NewClientUnix(conf ClientConfig) *Client {
@@ -138,11 +149,103 @@ func NewClientUnix(conf ClientConfig) *Client {
 			},
 		},
 	}
-
-	return &Client{
+	client := &Client{
 		server:    server,
 		requester: requester,
 	}
+	err = client.registerWorker()
+	if err != nil {
+		panic(err)
+	}
+	go client.workerHeartbeat()
+	return client
+}
+
+func (c *Client) registerWorker() error {
+	c.workerIDMu.Lock()
+	defer c.workerIDMu.Unlock()
+
+	url, err := c.server.Parse("workers")
+	if err != nil {
+		return err
+	}
+
+	var buf bytes.Buffer
+	err = json.NewEncoder(&buf).Encode(api.PostWorkersRequest{
+		Arch: common.CurrentArch(),
+	})
+	if err != nil {
+		logrus.Errorf("Unable create worker request: %v", err)
+		return err
+	}
+
+	resp, err := c.NewRequest("POST", url.String(), map[string]string{"Content-Type": "application/json"}, bytes.NewReader(buf.Bytes()))
+	if err != nil {
+		logrus.Errorf("Unable to register worker: %v", err)
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusCreated {
+		return errorFromResponse(resp, "error registering worker")
+	}
+
+	var wresp api.PostWorkersResponse
+	err = json.NewDecoder(resp.Body).Decode(&wresp)
+	if err != nil {
+		return err
+	}
+
+	workerID, err := uuid.Parse(wresp.WorkerId)
+	if err != nil {
+		return err
+	}
+	c.workerID = workerID
+	return nil
+}
+
+func (c *Client) workerHeartbeat() {
+	//nolint:staticcheck // avoid SA1015, this is an endless function
+	for range time.Tick(time.Minute * 1) {
+		workerID := func() uuid.UUID {
+			c.workerIDMu.RLock()
+			defer c.workerIDMu.RUnlock()
+			return c.workerID
+		}
+
+		if workerID() == uuid.Nil {
+			err := c.registerWorker()
+			if err != nil {
+				logrus.Errorf("Error registering worker, %v", err)
+				continue
+			}
+		}
+
+		url, err := c.server.Parse(fmt.Sprintf("workers/%s/status", workerID()))
+		if err != nil {
+			logrus.Errorf("Error parsing worker status: %v", err)
+			continue
+		}
+
+		var buf bytes.Buffer
+		resp, err := c.NewRequest("POST", url.String(), map[string]string{"Content-Type": "application/json"}, bytes.NewReader(buf.Bytes()))
+		if err != nil {
+			logrus.Errorf("Error updating worker status: %v", err)
+			continue
+		}
+		if resp.StatusCode == http.StatusBadRequest {
+			logrus.Warning("Registering worker again")
+			err = c.registerWorker()
+			if err != nil {
+				logrus.Errorf("Error registering worker, %v", err)
+			}
+			continue
+		}
+		if resp.StatusCode != http.StatusOK {
+			logrus.Errorf("Error updating worker status: %d", resp.StatusCode)
+			continue
+		}
+	}
 }
 
 func (c *Client) refreshAccessToken() error {
@@ -176,26 +279,41 @@ func (c *Client) refreshAccessToken() error {
 	return nil
 }
 
-func (c *Client) NewRequest(method, url string, headers map[string]string, body io.Reader) (*http.Response, error) {
+func (c *Client) NewRequest(method, url string, headers map[string]string, body io.ReadSeeker) (*http.Response, error) {
 	token := func() string {
 		c.tokenMu.RLock()
 		defer c.tokenMu.RUnlock()
 		return c.accessToken
 	}
 
-	req, err := http.NewRequest(method, url, body)
-	if err != nil {
-		return nil, err
-	}
+	newRequest := func() (*http.Request, error) {
+		if body != nil {
+			_, err := body.Seek(0, 0)
+			if err != nil {
+				return nil, err
+			}
+		}
 
-	if c.oAuthURL != "" {
-		req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token()))
-	}
+		req, err := http.NewRequest(method, url, body)
+		if err != nil {
+			return nil, err
+		}
 
-	for k, v := range headers {
-		req.Header.Add(k, v)
+		if c.oAuthURL != "" {
+			req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token()))
+		}
+
+		for k, v := range headers {
+			req.Header.Add(k, v)
+		}
+
+		return req, nil
 	}
 
+	req, err := newRequest()
+	if err != nil {
+		return nil, err
+	}
 	resp, err := c.requester.Do(req)
 	if err != nil {
 		return nil, err
@@ -207,7 +325,10 @@ func (c *Client) NewRequest(method, url string, headers map[string]string, body
 			return nil, err
 		}
 
-		req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", token()))
+		req, err = newRequest()
+		if err != nil {
+			return nil, err
+		}
 		resp, err = c.requester.Do(req)
 	}
 	return resp, err
@@ -220,16 +341,28 @@ func (c *Client) RequestJob(types []string, arch string) (Job, error) {
 		panic(err)
 	}
 
-	var buf bytes.Buffer
-	err = json.NewEncoder(&buf).Encode(api.RequestJobJSONRequestBody{
+	reqBody := api.RequestJobJSONRequestBody{
 		Types: types,
 		Arch:  arch,
-	})
+	}
+
+	workerID := func() uuid.UUID {
+		c.workerIDMu.RLock()
+		defer c.workerIDMu.RUnlock()
+		return c.workerID
+	}()
+
+	if workerID != uuid.Nil {
+		reqBody.WorkerId = common.ToPtr(workerID.String())
+	}
+
+	var buf bytes.Buffer
+	err = json.NewEncoder(&buf).Encode(reqBody)
 	if err != nil {
 		panic(err)
 	}
 
-	response, err := c.NewRequest("POST", url.String(), map[string]string{"Content-Type": "application/json"}, &buf)
+	response, err := c.NewRequest("POST", url.String(), map[string]string{"Content-Type": "application/json"}, bytes.NewReader(buf.Bytes()))
 	if err != nil {
 		return nil, err
 	}
@@ -320,7 +453,7 @@ func (j *job) Update(result interface{}) error {
 		panic(err)
 	}
 
-	response, err := j.client.NewRequest("PATCH", j.location, map[string]string{"Content-Type": "application/json"}, &buf)
+	response, err := j.client.NewRequest("PATCH", j.location, map[string]string{"Content-Type": "application/json"}, bytes.NewReader(buf.Bytes()))
 	if err != nil {
 		return fmt.Errorf("error fetching job info: %v", err)
 	}
@@ -353,7 +486,7 @@ func (j *job) Canceled() (bool, error) {
 	return jr.Canceled, nil
 }
 
-func (j *job) UploadArtifact(name string, reader io.Reader) error {
+func (j *job) UploadArtifact(name string, readSeeker io.ReadSeeker) error {
 	if j.artifactLocation == "" {
 		return fmt.Errorf("server does not accept artifacts for this job")
 	}
@@ -368,7 +501,7 @@ func (j *job) UploadArtifact(name string, reader io.Reader) error {
 		panic(err)
 	}
 
-	response, err := j.client.NewRequest("PUT", loc.String(), map[string]string{"Content-Type": "application/octet-stream"}, reader)
+	response, err := j.client.NewRequest("PUT", loc.String(), map[string]string{"Content-Type": "application/octet-stream"}, readSeeker)
 	if err != nil {
 		return fmt.Errorf("error uploading artifact: %v", err)
 	}
diff --git a/internal/worker/server.go b/internal/worker/server.go
index f746461fc..254c1421c 100644
--- a/internal/worker/server.go
+++ b/internal/worker/server.go
@@ -72,6 +72,8 @@ type Config struct {
 	BasePath             string
 	JWTEnabled           bool
 	TenantProviderFields []string
+	WorkerTimeout        time.Duration
+	WorkerWatchFreq      time.Duration
 }
 
 func NewServer(logger *log.Logger, jobs jobqueue.JobQueue, config Config) *Server {
@@ -81,9 +83,17 @@ func NewServer(logger *log.Logger, jobs jobqueue.JobQueue, config Config) *Serve
 		config: config,
 	}
 
+	if s.config.WorkerTimeout == 0 {
+		s.config.WorkerTimeout = time.Hour
+	}
+	if s.config.WorkerWatchFreq == 0 {
+		s.config.WorkerWatchFreq = time.Second * 300
+	}
+
 	api.BasePath = config.BasePath
 
 	go s.WatchHeartbeats()
+	go s.WatchWorkers()
 	return s
 }
 
@@ -143,6 +153,26 @@ func (s *Server) WatchHeartbeats() {
 	}
 }
 
+// This function should be started as a goroutine
+// Every 5 minutes it goes through all workers, removing any unresponsive ones.
+func (s *Server) WatchWorkers() {
+	//nolint:staticcheck // avoid SA1015, this is an endless function
+	for range time.Tick(s.config.WorkerWatchFreq) {
+		workers, err := s.jobs.Workers(s.config.WorkerTimeout)
+		if err != nil {
+			logrus.Warningf("Unable to query workers: %v", err)
+			continue
+		}
+		for _, wID := range workers {
+			logrus.Infof("Removing inactive worker: %s", wID)
+			err = s.jobs.DeleteWorker(wID)
+			if err != nil {
+				logrus.Warningf("Unable to remove worker: %v", err)
+			}
+		}
+	}
+}
+
 func (s *Server) EnqueueOSBuild(arch string, job *OSBuildJob, channel string) (uuid.UUID, error) {
 	return s.enqueue(JobTypeOSBuild+":"+arch, job, nil, channel)
 }
@@ -575,15 +605,15 @@ func (s *Server) DeleteArtifacts(id uuid.UUID) error {
 	return os.RemoveAll(path.Join(s.config.ArtifactsDir, id.String()))
 }
 
-func (s *Server) RequestJob(ctx context.Context, arch string, jobTypes []string, channels []string) (uuid.UUID, uuid.UUID, string, json.RawMessage, []json.RawMessage, error) {
-	return s.requestJob(ctx, arch, jobTypes, uuid.Nil, channels)
+func (s *Server) RequestJob(ctx context.Context, arch string, jobTypes, channels []string, workerID uuid.UUID) (uuid.UUID, uuid.UUID, string, json.RawMessage, []json.RawMessage, error) {
+	return s.requestJob(ctx, arch, jobTypes, uuid.Nil, channels, workerID)
 }
 
 func (s *Server) RequestJobById(ctx context.Context, arch string, requestedJobId uuid.UUID) (uuid.UUID, uuid.UUID, string, json.RawMessage, []json.RawMessage, error) {
-	return s.requestJob(ctx, arch, []string{}, requestedJobId, nil)
+	return s.requestJob(ctx, arch, []string{}, requestedJobId, nil, uuid.Nil)
 }
 
-func (s *Server) requestJob(ctx context.Context, arch string, jobTypes []string, requestedJobId uuid.UUID, channels []string) (
+func (s *Server) requestJob(ctx context.Context, arch string, jobTypes []string, requestedJobId uuid.UUID, channels []string, workerID uuid.UUID) (
 	jobId uuid.UUID, token uuid.UUID, jobType string, args json.RawMessage, dynamicArgs []json.RawMessage, err error) {
 	// treat osbuild jobs specially until we have found a generic way to
 	// specify dequeuing restrictions. For now, we only have one
@@ -614,9 +644,9 @@ func (s *Server) requestJob(ctx context.Context, arch string, jobTypes []string,
 	var depIDs []uuid.UUID
 	if requestedJobId != uuid.Nil {
 		jobId = requestedJobId
-		token, depIDs, jobType, args, err = s.jobs.DequeueByID(dequeueCtx, requestedJobId)
+		token, depIDs, jobType, args, err = s.jobs.DequeueByID(dequeueCtx, requestedJobId, workerID)
 	} else {
-		jobId, token, depIDs, jobType, args, err = s.jobs.Dequeue(dequeueCtx, jts, channels)
+		jobId, token, depIDs, jobType, args, err = s.jobs.Dequeue(dequeueCtx, workerID, jts, channels)
 	}
 	if err != nil {
 		if err != jobqueue.ErrDequeueTimeout && err != jobqueue.ErrNotPending {
@@ -852,7 +882,15 @@ func (h *apiHandlers) RequestJob(ctx echo.Context) error {
 		channel = "org-" + tenant
 	}
 
-	jobId, jobToken, jobType, jobArgs, dynamicJobArgs, err := h.server.RequestJob(ctx.Request().Context(), body.Arch, body.Types, []string{channel})
+	workerID := uuid.Nil
+	if body.WorkerId != nil {
+		workerID, err = uuid.Parse(*body.WorkerId)
+		if err != nil {
+			return api.HTTPErrorWithInternal(api.ErrorMalformedWorkerId, err)
+		}
+	}
+
+	jobId, jobToken, jobType, jobArgs, dynamicJobArgs, err := h.server.RequestJob(ctx.Request().Context(), body.Arch, body.Types, []string{channel}, workerID)
 	if err != nil {
 		if err == jobqueue.ErrDequeueTimeout {
 			return ctx.JSON(http.StatusNoContent, api.ObjectReference{
@@ -992,6 +1030,47 @@ func (h *apiHandlers) UploadJobArtifact(ctx echo.Context, tokenstr string, name
 	return ctx.NoContent(http.StatusOK)
 }
 
+func (h *apiHandlers) PostWorkers(ctx echo.Context) error {
+	var body api.PostWorkersRequest
+	err := ctx.Bind(&body)
+	if err != nil {
+		return err
+	}
+
+	workerID, err := h.server.jobs.InsertWorker(body.Arch)
+	if err != nil {
+		return api.HTTPErrorWithInternal(api.ErrorInsertingWorker, err)
+	}
+	logrus.Infof("Worker (%v) registered", body.Arch)
+
+	return ctx.JSON(http.StatusCreated, api.PostWorkersResponse{
+		ObjectReference: api.ObjectReference{
+			Href: fmt.Sprintf("%s/workers", api.BasePath),
+			Id:   workerID.String(),
+			Kind: "WorkerID",
+		},
+		WorkerId: workerID.String(),
+	})
+}
+
+func (h *apiHandlers) PostWorkerStatus(ctx echo.Context, workerIdstr string) error {
+	workerID, err := uuid.Parse(workerIdstr)
+	if err != nil {
+		return api.HTTPErrorWithInternal(api.ErrorMalformedWorkerId, err)
+	}
+	err = h.server.jobs.UpdateWorkerStatus(workerID)
+
+	if err == jobqueue.ErrWorkerNotExist {
+		return api.HTTPErrorWithInternal(api.ErrorWorkerIdNotFound, err)
+	}
+
+	if err != nil {
+		return api.HTTPErrorWithInternal(api.ErrorUpdatingWorkerStatus, err)
+	}
+
+	return ctx.NoContent(http.StatusOK)
+}
+
 // A simple echo.Binder(), which only accepts application/json, but is more
 // strict than echo's DefaultBinder. It does not handle binding query
 // parameters either.
diff --git a/pkg/jobqueue/dbjobqueue/dbjobqueue.go b/pkg/jobqueue/dbjobqueue/dbjobqueue.go
index 8437301a9..fc56dbb46 100644
--- a/pkg/jobqueue/dbjobqueue/dbjobqueue.go
+++ b/pkg/jobqueue/dbjobqueue/dbjobqueue.go
@@ -95,19 +95,41 @@ const (
 		RETURNING type, started_at`
 
 	sqlInsertHeartbeat = `
-                INSERT INTO heartbeats(token, id, heartbeat)
-                VALUES ($1, $2, now())`
+		INSERT INTO heartbeats(token, id, heartbeat)
+		VALUES ($1, $2, now())`
+	sqlInsertHeartbeatWithWorker = `
+		INSERT INTO heartbeats(token, id, worker_id, heartbeat)
+		VALUES ($1, $2, $3, now())`
 	sqlQueryHeartbeats = `
-                SELECT token
-                FROM heartbeats
-                WHERE age(now(), heartbeat) > $1`
+		SELECT token
+		FROM heartbeats
+		WHERE age(now(), heartbeat) > $1`
 	sqlRefreshHeartbeat = `
-                UPDATE heartbeats
-                SET heartbeat = now()
-                WHERE token = $1`
+		UPDATE heartbeats
+		SET heartbeat = now()
+		WHERE token = $1`
 	sqlDeleteHeartbeat = `
-                DELETE FROM heartbeats
-                WHERE id = $1`
+		DELETE FROM heartbeats
+		WHERE id = $1`
+	sqlQueryHeartbeatsForWorker = `
+		SELECT count(token)
+		FROM heartbeats
+		WHERE worker_id = $1`
+
+	sqlInsertWorker = `
+		INSERT INTO workers(worker_id, arch, heartbeat)
+		VALUES($1, $2, now())`
+	sqlUpdateWorkerStatus = `
+		UPDATE heartbeats
+		SET heartbeat = now()
+		WHERE worker_id = $1`
+	sqlQueryWorkers = `
+		SELECT worker_id
+		FROM workers
+		WHERE age(now(), heartbeat) > $1`
+	sqlDeleteWorker = `
+		DELETE FROM workers
+		WHERE worker_id = $1`
 )
 
 type DBJobQueue struct {
@@ -307,7 +329,7 @@ func (q *DBJobQueue) Enqueue(jobType string, args interface{}, dependencies []uu
 	return id, nil
 }
 
-func (q *DBJobQueue) Dequeue(ctx context.Context, jobTypes []string, channels []string) (uuid.UUID, uuid.UUID, []uuid.UUID, string, json.RawMessage, error) {
+func (q *DBJobQueue) Dequeue(ctx context.Context, workerID uuid.UUID, jobTypes, channels []string) (uuid.UUID, uuid.UUID, []uuid.UUID, string, json.RawMessage, error) {
 	// add ourselves as a dequeuer
 	c := make(chan struct{}, 1)
 	el := q.dequeuers.pushBack(c)
@@ -316,7 +338,7 @@ func (q *DBJobQueue) Dequeue(ctx context.Context, jobTypes []string, channels []
 	token := uuid.New()
 	for {
 		var err error
-		id, dependencies, jobType, args, err := q.dequeueMaybe(ctx, token, jobTypes, channels)
+		id, dependencies, jobType, args, err := q.dequeueMaybe(ctx, token, workerID, jobTypes, channels)
 		if err == nil {
 			return id, token, dependencies, jobType, args, nil
 		}
@@ -343,7 +365,7 @@ func (q *DBJobQueue) Dequeue(ctx context.Context, jobTypes []string, channels []
 //
 // This method returns pgx.ErrNoRows if the method didn't manage to dequeue
 // anything
-func (q *DBJobQueue) dequeueMaybe(ctx context.Context, token uuid.UUID, jobTypes []string, channels []string) (uuid.UUID, []uuid.UUID, string, json.RawMessage, error) {
+func (q *DBJobQueue) dequeueMaybe(ctx context.Context, token, workerID uuid.UUID, jobTypes, channels []string) (uuid.UUID, []uuid.UUID, string, json.RawMessage, error) {
 	var conn *pgxpool.Conn
 	conn, err := q.pool.Acquire(ctx)
 	if err != nil {
@@ -374,9 +396,16 @@ func (q *DBJobQueue) dequeueMaybe(ctx context.Context, token uuid.UUID, jobTypes
 	}
 
 	// insert heartbeat
-	_, err = tx.Exec(ctx, sqlInsertHeartbeat, token, id)
-	if err != nil {
-		return uuid.Nil, nil, "", nil, fmt.Errorf("error inserting the job's heartbeat: %v", err)
+	if workerID != uuid.Nil {
+		_, err = tx.Exec(ctx, sqlInsertHeartbeatWithWorker, token, id, workerID)
+		if err != nil {
+			return uuid.Nil, nil, "", nil, fmt.Errorf("error inserting the job's heartbeat: %v", err)
+		}
+	} else {
+		_, err = tx.Exec(ctx, sqlInsertHeartbeat, token, id)
+		if err != nil {
+			return uuid.Nil, nil, "", nil, fmt.Errorf("error inserting the job's heartbeat: %v", err)
+		}
 	}
 
 	dependencies, err := q.jobDependencies(ctx, tx, id)
@@ -394,7 +423,7 @@ func (q *DBJobQueue) dequeueMaybe(ctx context.Context, token uuid.UUID, jobTypes
 	return id, dependencies, jobType, args, nil
 }
 
-func (q *DBJobQueue) DequeueByID(ctx context.Context, id uuid.UUID) (uuid.UUID, []uuid.UUID, string, json.RawMessage, error) {
+func (q *DBJobQueue) DequeueByID(ctx context.Context, id, workerID uuid.UUID) (uuid.UUID, []uuid.UUID, string, json.RawMessage, error) {
 	// Return early if the context is already canceled.
 	if err := ctx.Err(); err != nil {
 		return uuid.Nil, nil, "", nil, jobqueue.ErrDequeueTimeout
@@ -431,9 +460,16 @@ func (q *DBJobQueue) DequeueByID(ctx context.Context, id uuid.UUID) (uuid.UUID,
 	}
 
 	// insert heartbeat
-	_, err = tx.Exec(ctx, sqlInsertHeartbeat, token, id)
-	if err != nil {
-		return uuid.Nil, nil, "", nil, fmt.Errorf("error inserting the job's heartbeat: %v", err)
+	if workerID != uuid.Nil {
+		_, err = tx.Exec(ctx, sqlInsertHeartbeatWithWorker, token, id, workerID)
+		if err != nil {
+			return uuid.Nil, nil, "", nil, fmt.Errorf("error inserting the job's heartbeat: %v", err)
+		}
+	} else {
+		_, err = tx.Exec(ctx, sqlInsertHeartbeat, token, id)
+		if err != nil {
+			return uuid.Nil, nil, "", nil, fmt.Errorf("error inserting the job's heartbeat: %v", err)
+		}
 	}
 
 	dependencies, err := q.jobDependencies(ctx, tx, id)
@@ -486,7 +522,7 @@ func (q *DBJobQueue) RequeueOrFinishJob(id uuid.UUID, maxRetries uint64, result
 		return jobqueue.ErrNotRunning
 	}
 
-	// Remove from heartbeats
+	// Remove from heartbeats if token is null
 	tag, err := tx.Exec(context.Background(), sqlDeleteHeartbeat, id)
 	if err != nil {
 		return fmt.Errorf("error removing job %s from heartbeats: %v", id, err)
@@ -677,6 +713,102 @@ func (q *DBJobQueue) RefreshHeartbeat(token uuid.UUID) {
 	}
 }
 
+func (q *DBJobQueue) InsertWorker(arch string) (uuid.UUID, error) {
+	conn, err := q.pool.Acquire(context.Background())
+	if err != nil {
+		return uuid.Nil, err
+	}
+	defer conn.Release()
+
+	id := uuid.New()
+	_, err = conn.Exec(context.Background(), sqlInsertWorker, id, arch)
+	if err != nil {
+		q.logger.Error(err, "Error inserting worker")
+		return uuid.Nil, err
+	}
+	return id, nil
+}
+
+func (q *DBJobQueue) UpdateWorkerStatus(workerID uuid.UUID) error {
+	conn, err := q.pool.Acquire(context.Background())
+	if err != nil {
+		return err
+	}
+	defer conn.Release()
+
+	tag, err := conn.Exec(context.Background(), sqlUpdateWorkerStatus, workerID)
+	if err != nil {
+		q.logger.Error(err, "Error updating  worker status")
+	}
+	if tag.RowsAffected() != 1 {
+		q.logger.Error(nil, "No rows affected when refreshing updating status")
+		return jobqueue.ErrWorkerNotExist
+	}
+	return nil
+}
+func (q *DBJobQueue) Workers(olderThan time.Duration) ([]uuid.UUID, error) {
+	conn, err := q.pool.Acquire(context.Background())
+	if err != nil {
+		return nil, err
+	}
+	defer conn.Release()
+
+	rows, err := conn.Query(context.Background(), sqlQueryWorkers, olderThan.String())
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	workerIDs := make([]uuid.UUID, 0)
+	for rows.Next() {
+		var w uuid.UUID
+		err = rows.Scan(&w)
+		if err != nil {
+			// Log the error and try to continue with the next row
+			q.logger.Error(err, "Unable to read token from heartbeats")
+			continue
+		}
+		workerIDs = append(workerIDs, w)
+	}
+	if rows.Err() != nil {
+		q.logger.Error(rows.Err(), "Error reading tokens from heartbeats")
+		return nil, rows.Err()
+	}
+
+	return workerIDs, nil
+}
+
+func (q *DBJobQueue) DeleteWorker(workerID uuid.UUID) error {
+	conn, err := q.pool.Acquire(context.Background())
+	if err != nil {
+		return err
+	}
+	defer conn.Release()
+
+	var count int
+	err = conn.QueryRow(context.Background(), sqlQueryHeartbeatsForWorker, workerID).Scan(&count)
+	if err != nil {
+		return err
+	}
+
+	// If worker has any active jobs, refuse to remove the worker
+	if count != 0 {
+		return jobqueue.ErrActiveJobs
+	}
+
+	tag, err := conn.Exec(context.Background(), sqlDeleteWorker, workerID)
+	if err != nil {
+		q.logger.Error(err, "Error deleting worker")
+		return err
+	}
+
+	if tag.RowsAffected() != 1 {
+		q.logger.Error(nil, "No rows affected when deleting worker")
+		return jobqueue.ErrActiveJobs
+	}
+	return nil
+}
+
 // connection unifies pgxpool.Conn and pgx.Tx interfaces
 // Some methods don't care whether they run queries on a raw connection,
 // or in a transaction. This interface thus abstracts this concept.
diff --git a/pkg/jobqueue/dbjobqueue/schemas/007_create_table_workers.sql b/pkg/jobqueue/dbjobqueue/schemas/007_create_table_workers.sql
new file mode 100644
index 000000000..4b1028857
--- /dev/null
+++ b/pkg/jobqueue/dbjobqueue/schemas/007_create_table_workers.sql
@@ -0,0 +1,8 @@
+CREATE TABLE workers(
+       worker_id uuid PRIMARY KEY,
+       arch varchar NOT NULL,
+       heartbeat timestamp NOT NULL
+);
+
+ALTER TABLE heartbeats
+ADD COLUMN worker_id uuid REFERENCES workers(worker_id) ON DELETE CASCADE;
diff --git a/pkg/jobqueue/jobqueue.go b/pkg/jobqueue/jobqueue.go
index 8b11f5f86..55c357cab 100644
--- a/pkg/jobqueue/jobqueue.go
+++ b/pkg/jobqueue/jobqueue.go
@@ -41,13 +41,13 @@ type JobQueue interface {
 	//
 	// Returns the job's id, token, dependencies, type, and arguments, or an error. Arguments
 	// can be unmarshaled to the type given in Enqueue().
-	Dequeue(ctx context.Context, jobTypes []string, channels []string) (uuid.UUID, uuid.UUID, []uuid.UUID, string, json.RawMessage, error)
+	Dequeue(ctx context.Context, workerID uuid.UUID, jobTypes []string, channels []string) (uuid.UUID, uuid.UUID, []uuid.UUID, string, json.RawMessage, error)
 
 	// Dequeues a pending job by its ID in a non-blocking way.
 	//
 	// Returns the job's token, dependencies, type, and arguments, or an error. Arguments
 	// can be unmarshaled to the type given in Enqueue().
-	DequeueByID(ctx context.Context, id uuid.UUID) (uuid.UUID, []uuid.UUID, string, json.RawMessage, error)
+	DequeueByID(ctx context.Context, id, workerID uuid.UUID) (uuid.UUID, []uuid.UUID, string, json.RawMessage, error)
 
 	// Tries to requeue a running job by its ID
 	//
@@ -78,8 +78,20 @@ type JobQueue interface {
 	// Get a list of tokens which haven't been updated in the specified time frame
 	Heartbeats(olderThan time.Duration) (tokens []uuid.UUID)
 
-	// Reset the last heartbeat time to time.Now()
+	// Reset the last job heartbeat time to time.Now()
 	RefreshHeartbeat(token uuid.UUID)
+
+	// Inserts the worker and creates a UUID for it
+	InsertWorker(arch string) (uuid.UUID, error)
+
+	// Reset the last worker's heartbeat time to time.Now()
+	UpdateWorkerStatus(workerID uuid.UUID) error
+
+	// Get a list of workers which haven't been updated in the specified time frame
+	Workers(olderThan time.Duration) ([]uuid.UUID, error)
+
+	// Deletes the worker
+	DeleteWorker(workerID uuid.UUID) error
 }
 
 // SimpleLogger provides a structured logging methods for the jobqueue library.
@@ -100,4 +112,6 @@ var (
 	ErrNotRunning     = errors.New("job is not running")
 	ErrCanceled       = errors.New("job was canceled")
 	ErrDequeueTimeout = errors.New("dequeue context timed out or was canceled")
+	ErrActiveJobs     = errors.New("worker has active jobs associated with it")
+	ErrWorkerNotExist = errors.New("worker does not exist")
 )
diff --git a/schutzbot/terraform b/schutzbot/terraform
index 46ea0a6f3..b1c3b3d02 100644
--- a/schutzbot/terraform
+++ b/schutzbot/terraform
@@ -1,1 +1,1 @@
-93c6080642ad10add2fddbf847e95ef7f42ee48c
+f127f38ebe23ef9a6ef4e6dcaf6d5cd9389f0a08


Test patch:
diff --git a/cmd/osbuild-service-maintenance/db_test.go b/cmd/osbuild-service-maintenance/db_test.go
index ec45a7112..517645320 100644
--- a/cmd/osbuild-service-maintenance/db_test.go
+++ b/cmd/osbuild-service-maintenance/db_test.go
@@ -44,7 +44,7 @@ func testDeleteJob(t *testing.T, d db, q *dbjobqueue.DBJobQueue) {
 	id, err := q.Enqueue("octopus", nil, nil, "")
 	require.NoError(t, err)
 	require.NotEqual(t, uuid.Nil, id)
-	_, _, _, _, _, err = q.Dequeue(context.Background(), []string{"octopus"}, []string{""})
+	_, _, _, _, _, err = q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{""})
 	require.NoError(t, err)
 
 	type Result struct {
diff --git a/internal/cloudapi/v2/v2_koji_test.go b/internal/cloudapi/v2/v2_koji_test.go
index 9d1611277..7a2fa3c34 100644
--- a/internal/cloudapi/v2/v2_koji_test.go
+++ b/internal/cloudapi/v2/v2_koji_test.go
@@ -418,7 +418,7 @@ func TestKojiCompose(t *testing.T) {
 			require.NoError(t, err)
 
 			// handle koji-init
-			_, token, jobType, rawJob, _, err := workerServer.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeKojiInit}, []string{""})
+			_, token, jobType, rawJob, _, err := workerServer.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeKojiInit}, []string{""}, uuid.Nil)
 			require.NoError(t, err)
 			require.Equal(t, worker.JobTypeKojiInit, jobType)
 
@@ -445,7 +445,7 @@ func TestKojiCompose(t *testing.T) {
 
 			// handle build jobs
 			for i := 0; i < len(buildJobIDs); i++ {
-				jobID, token, jobType, rawJob, _, err := workerServer.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""})
+				jobID, token, jobType, rawJob, _, err := workerServer.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 				require.NoError(t, err)
 				require.Equal(t, worker.JobTypeOSBuild, jobType)
 
@@ -485,7 +485,7 @@ func TestKojiCompose(t *testing.T) {
 			}
 
 			// handle koji-finalize
-			finalizeID, token, jobType, rawJob, _, err := workerServer.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeKojiFinalize}, []string{""})
+			finalizeID, token, jobType, rawJob, _, err := workerServer.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeKojiFinalize}, []string{""}, uuid.Nil)
 			require.NoError(t, err)
 			require.Equal(t, worker.JobTypeKojiFinalize, jobType)
 
diff --git a/internal/cloudapi/v2/v2_test.go b/internal/cloudapi/v2/v2_test.go
index 439eaf886..32c1cfc95 100644
--- a/internal/cloudapi/v2/v2_test.go
+++ b/internal/cloudapi/v2/v2_test.go
@@ -8,6 +8,7 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 
 	"github.com/osbuild/osbuild-composer/pkg/jobqueue"
@@ -49,7 +50,7 @@ func newV2Server(t *testing.T, dir string, depsolveChannels []string, enableJWT
 	go func() {
 		defer wg.Done()
 		for {
-			_, token, _, _, _, err := workerServer.RequestJob(depsolveContext, test_distro.TestDistroName, []string{worker.JobTypeDepsolve}, depsolveChannels)
+			_, token, _, _, _, err := workerServer.RequestJob(depsolveContext, test_distro.TestDistroName, []string{worker.JobTypeDepsolve}, depsolveChannels, uuid.Nil)
 			select {
 			case <-depsolveContext.Done():
 				return
@@ -86,7 +87,7 @@ func newV2Server(t *testing.T, dir string, depsolveChannels []string, enableJWT
 	go func() {
 		defer wg.Done()
 		for {
-			_, token, _, _, _, err := workerServer.RequestJob(ostreeResolveContext, test_distro.TestDistroName, []string{worker.JobTypeOSTreeResolve}, depsolveChannels)
+			_, token, _, _, _, err := workerServer.RequestJob(ostreeResolveContext, test_distro.TestDistroName, []string{worker.JobTypeOSTreeResolve}, depsolveChannels, uuid.Nil)
 			select {
 			case <-ostreeResolveContext.Done():
 				return
@@ -597,7 +598,7 @@ func TestComposeStatusSuccess(t *testing.T) {
 		"kind": "ComposeId"
 	}`, "id")
 
-	jobId, token, jobType, args, dynArgs, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""})
+	jobId, token, jobType, args, dynArgs, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, worker.JobTypeOSBuild, jobType)
 
@@ -708,7 +709,7 @@ func TestComposeStatusFailure(t *testing.T) {
 		"kind": "ComposeId"
 	}`, "id")
 
-	jobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""})
+	jobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, worker.JobTypeOSBuild, jobType)
 
@@ -779,7 +780,7 @@ func TestComposeJobError(t *testing.T) {
 		"kind": "ComposeId"
 	}`, "id")
 
-	jobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""})
+	jobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, worker.JobTypeOSBuild, jobType)
 
@@ -844,7 +845,7 @@ func TestComposeDependencyError(t *testing.T) {
 		"kind": "ComposeId"
 	}`, "id")
 
-	jobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""})
+	jobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, worker.JobTypeOSBuild, jobType)
 
@@ -917,7 +918,7 @@ func TestComposeTargetErrors(t *testing.T) {
 		"kind": "ComposeId"
 	}`, "id")
 
-	jobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""})
+	jobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, worker.JobTypeOSBuild, jobType)
 
@@ -1337,7 +1338,7 @@ func TestImageFromCompose(t *testing.T) {
 		"kind": "ComposeId"
 	}`, "id")
 
-	jobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""})
+	jobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, worker.JobTypeOSBuild, jobType)
 
@@ -1406,7 +1407,7 @@ func TestImageFromCompose(t *testing.T) {
 		"kind": "CloneComposeId"
 	}`, jobId), "id")
 
-	_, token, jobType, _, _, err = wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeAWSEC2Copy}, []string{""})
+	_, token, jobType, _, _, err = wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeAWSEC2Copy}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, worker.JobTypeAWSEC2Copy, jobType)
 
@@ -1418,7 +1419,7 @@ func TestImageFromCompose(t *testing.T) {
 	err = wrksrv.FinishJob(token, res)
 	require.NoError(t, err)
 
-	imgJobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeAWSEC2Share}, []string{""})
+	imgJobId, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeAWSEC2Share}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, worker.JobTypeAWSEC2Share, jobType)
 
diff --git a/internal/jobqueue/jobqueuetest/jobqueuetest.go b/internal/jobqueue/jobqueuetest/jobqueuetest.go
index fa2d688e8..a558bd960 100644
--- a/internal/jobqueue/jobqueuetest/jobqueuetest.go
+++ b/internal/jobqueue/jobqueuetest/jobqueuetest.go
@@ -44,6 +44,7 @@ func TestJobQueue(t *testing.T, makeJobQueue MakeJobQueue) {
 	t.Run("dequeue-by-id", wrap(testDequeueByID))
 	t.Run("multiple-channels", wrap(testMultipleChannels))
 	t.Run("100-dequeuers", wrap(test100dequeuers))
+	t.Run("workers", wrap(testWorkers))
 }
 
 func pushTestJob(t *testing.T, q jobqueue.JobQueue, jobType string, args interface{}, dependencies []uuid.UUID, channel string) uuid.UUID {
@@ -55,7 +56,7 @@ func pushTestJob(t *testing.T, q jobqueue.JobQueue, jobType string, args interfa
 }
 
 func finishNextTestJob(t *testing.T, q jobqueue.JobQueue, jobType string, result interface{}, deps []uuid.UUID) uuid.UUID {
-	id, tok, d, typ, args, err := q.Dequeue(context.Background(), []string{jobType}, []string{""})
+	id, tok, d, typ, args, err := q.Dequeue(context.Background(), uuid.Nil, []string{jobType}, []string{""})
 	require.NoError(t, err)
 	require.NotEmpty(t, id)
 	require.NotEmpty(t, tok)
@@ -82,7 +83,7 @@ func testErrors(t *testing.T, q jobqueue.JobQueue) {
 
 	// token gets removed
 	pushTestJob(t, q, "octopus", nil, nil, "")
-	id, tok, _, _, _, err := q.Dequeue(context.Background(), []string{"octopus"}, []string{""})
+	id, tok, _, _, _, err := q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{""})
 	require.NoError(t, err)
 	require.NotEmpty(t, tok)
 
@@ -133,7 +134,7 @@ func testArgs(t *testing.T, q jobqueue.JobQueue) {
 	require.Equal(t, "kingfisher", jchan)
 	require.Equal(t, "octopus", jtype)
 
-	id, tok, deps, typ, args, err := q.Dequeue(context.Background(), []string{"octopus"}, []string{"kingfisher"})
+	id, tok, deps, typ, args, err := q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{"kingfisher"})
 	require.NoError(t, err)
 	require.Equal(t, two, id)
 	require.NotEmpty(t, tok)
@@ -151,7 +152,7 @@ func testArgs(t *testing.T, q jobqueue.JobQueue) {
 	require.Equal(t, "kingfisher", jchan)
 	require.Equal(t, typ, jtype)
 
-	id, tok, deps, typ, args, err = q.Dequeue(context.Background(), []string{"fish"}, []string{"toucan"})
+	id, tok, deps, typ, args, err = q.Dequeue(context.Background(), uuid.Nil, []string{"fish"}, []string{"toucan"})
 	require.NoError(t, err)
 	require.Equal(t, one, id)
 	require.NotEmpty(t, tok)
@@ -181,7 +182,7 @@ func testJobTypes(t *testing.T, q jobqueue.JobQueue) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
-	id, tok, deps, typ, args, err := q.Dequeue(ctx, []string{"zebra"}, []string{""})
+	id, tok, deps, typ, args, err := q.Dequeue(ctx, uuid.Nil, []string{"zebra"}, []string{""})
 	require.Equal(t, err, jobqueue.ErrDequeueTimeout)
 	require.Equal(t, uuid.Nil, id)
 	require.Equal(t, uuid.Nil, tok)
@@ -193,12 +194,12 @@ func testJobTypes(t *testing.T, q jobqueue.JobQueue) {
 func testDequeueTimeout(t *testing.T, q jobqueue.JobQueue) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*20)
 	defer cancel()
-	_, _, _, _, _, err := q.Dequeue(ctx, []string{"octopus"}, []string{""})
+	_, _, _, _, _, err := q.Dequeue(ctx, uuid.Nil, []string{"octopus"}, []string{""})
 	require.Equal(t, jobqueue.ErrDequeueTimeout, err)
 
 	ctx2, cancel2 := context.WithCancel(context.Background())
 	cancel2()
-	_, _, _, _, _, err = q.Dequeue(ctx2, []string{"octopus"}, []string{""})
+	_, _, _, _, _, err = q.Dequeue(ctx2, uuid.Nil, []string{"octopus"}, []string{""})
 	require.Equal(t, jobqueue.ErrDequeueTimeout, err)
 }
 
@@ -286,7 +287,7 @@ func testMultipleWorkers(t *testing.T, q jobqueue.JobQueue) {
 		defer close(done)
 		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 		defer cancel()
-		id, tok, deps, typ, args, err := q.Dequeue(ctx, []string{"octopus"}, []string{""})
+		id, tok, deps, typ, args, err := q.Dequeue(ctx, uuid.Nil, []string{"octopus"}, []string{""})
 		require.NoError(t, err)
 		require.NotEmpty(t, id)
 		require.NotEmpty(t, tok)
@@ -301,7 +302,7 @@ func testMultipleWorkers(t *testing.T, q jobqueue.JobQueue) {
 
 	// This call to Dequeue() should not block on the one in the goroutine.
 	id := pushTestJob(t, q, "clownfish", nil, nil, "")
-	r, tok, deps, typ, args, err := q.Dequeue(context.Background(), []string{"clownfish"}, []string{""})
+	r, tok, deps, typ, args, err := q.Dequeue(context.Background(), uuid.Nil, []string{"clownfish"}, []string{""})
 	require.NoError(t, err)
 	require.Equal(t, id, r)
 	require.NotEmpty(t, tok)
@@ -324,7 +325,7 @@ func testMultipleWorkersSingleJobType(t *testing.T, q jobqueue.JobQueue) {
 			defer wg.Add(-1)
 			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 			defer cancel()
-			id, tok, deps, typ, args, err := q.Dequeue(ctx, []string{"clownfish"}, []string{""})
+			id, tok, deps, typ, args, err := q.Dequeue(ctx, uuid.Nil, []string{"clownfish"}, []string{""})
 			require.NoError(t, err)
 			require.NotEmpty(t, id)
 			require.NotEmpty(t, tok)
@@ -371,7 +372,7 @@ func testCancel(t *testing.T, q jobqueue.JobQueue) {
 	// 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, tok, deps, typ, args, err := q.Dequeue(context.Background(), []string{"clownfish"}, []string{""})
+	r, tok, deps, typ, args, err := q.Dequeue(context.Background(), uuid.Nil, []string{"clownfish"}, []string{""})
 	require.NoError(t, err)
 	require.Equal(t, id, r)
 	require.NotEmpty(t, tok)
@@ -391,7 +392,7 @@ func testCancel(t *testing.T, q jobqueue.JobQueue) {
 	// Cancel a finished job, which is a no-op
 	id = pushTestJob(t, q, "clownfish", nil, nil, "")
 	require.NotEmpty(t, id)
-	r, tok, deps, typ, args, err = q.Dequeue(context.Background(), []string{"clownfish"}, []string{""})
+	r, tok, deps, typ, args, err = q.Dequeue(context.Background(), uuid.Nil, []string{"clownfish"}, []string{""})
 	require.NoError(t, err)
 	require.Equal(t, id, r)
 	require.NotEmpty(t, tok)
@@ -423,7 +424,7 @@ func testRequeue(t *testing.T, q jobqueue.JobQueue) {
 	require.Error(t, err)
 
 	// Requeue a running job
-	r, tok1, deps, typ, args, err := q.Dequeue(context.Background(), []string{"clownfish"}, []string{""})
+	r, tok1, deps, typ, args, err := q.Dequeue(context.Background(), uuid.Nil, []string{"clownfish"}, []string{""})
 	require.NoError(t, err)
 	require.Equal(t, id, r)
 	require.NotEmpty(t, tok1)
@@ -432,7 +433,7 @@ func testRequeue(t *testing.T, q jobqueue.JobQueue) {
 	require.Equal(t, json.RawMessage("null"), args)
 	err = q.RequeueOrFinishJob(id, 1, nil)
 	require.NoError(t, err)
-	r, tok2, deps, typ, args, err := q.Dequeue(context.Background(), []string{"clownfish"}, []string{""})
+	r, tok2, deps, typ, args, err := q.Dequeue(context.Background(), uuid.Nil, []string{"clownfish"}, []string{""})
 	require.NoError(t, err)
 	require.Equal(t, id, r)
 	require.NotEmpty(t, tok2)
@@ -457,13 +458,13 @@ func testRequeueLimit(t *testing.T, q jobqueue.JobQueue) {
 	// Start a job
 	id := pushTestJob(t, q, "clownfish", nil, nil, "")
 	require.NotEmpty(t, id)
-	_, _, _, _, _, err := q.Dequeue(context.Background(), []string{"clownfish"}, []string{""})
+	_, _, _, _, _, err := q.Dequeue(context.Background(), uuid.Nil, []string{"clownfish"}, []string{""})
 	require.NoError(t, err)
 	// Requeue once
 	err = q.RequeueOrFinishJob(id, 1, nil)
 	require.NoError(t, err)
 	// Start again
-	_, _, _, _, _, err = q.Dequeue(context.Background(), []string{"clownfish"}, []string{""})
+	_, _, _, _, _, err = q.Dequeue(context.Background(), uuid.Nil, []string{"clownfish"}, []string{""})
 	require.NoError(t, err)
 	_, _, result, _, _, finished, _, _, _, err := q.JobStatus(id)
 	require.NoError(t, err)
@@ -483,7 +484,7 @@ func testHeartbeats(t *testing.T, q jobqueue.JobQueue) {
 	// No heartbeats for queued job
 	require.Empty(t, q.Heartbeats(time.Second*0))
 
-	r, tok, _, _, _, err := q.Dequeue(context.Background(), []string{"octopus"}, []string{""})
+	r, tok, _, _, _, err := q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{""})
 	require.NoError(t, err)
 	require.Equal(t, id, r)
 	require.NotEmpty(t, tok)
@@ -518,7 +519,7 @@ func testDequeueByID(t *testing.T, q jobqueue.JobQueue) {
 		one := pushTestJob(t, q, "octopus", nil, nil, "")
 		two := pushTestJob(t, q, "octopus", nil, nil, "")
 
-		tok, d, typ, args, err := q.DequeueByID(context.Background(), one)
+		tok, d, typ, args, err := q.DequeueByID(context.Background(), one, uuid.Nil)
 		require.NoError(t, err)
 		require.NotEmpty(t, tok)
 		require.Empty(t, d)
@@ -535,7 +536,7 @@ func testDequeueByID(t *testing.T, q jobqueue.JobQueue) {
 		one := pushTestJob(t, q, "octopus", nil, nil, "")
 		two := pushTestJob(t, q, "octopus", nil, []uuid.UUID{one}, "")
 
-		_, _, _, _, err := q.DequeueByID(context.Background(), two)
+		_, _, _, _, err := q.DequeueByID(context.Background(), two, uuid.Nil)
 		require.Equal(t, jobqueue.ErrNotPending, err)
 
 		require.Equal(t, one, finishNextTestJob(t, q, "octopus", testResult{}, nil))
@@ -545,16 +546,16 @@ func testDequeueByID(t *testing.T, q jobqueue.JobQueue) {
 	t.Run("cannot dequeue a non-pending job", func(t *testing.T) {
 		one := pushTestJob(t, q, "octopus", nil, nil, "")
 
-		_, _, _, _, _, err := q.Dequeue(context.Background(), []string{"octopus"}, []string{""})
+		_, _, _, _, _, err := q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{""})
 		require.NoError(t, err)
 
-		_, _, _, _, err = q.DequeueByID(context.Background(), one)
+		_, _, _, _, err = q.DequeueByID(context.Background(), one, uuid.Nil)
 		require.Equal(t, jobqueue.ErrNotPending, err)
 
 		err = q.RequeueOrFinishJob(one, 0, nil)
 		require.NoError(t, err)
 
-		_, _, _, _, err = q.DequeueByID(context.Background(), one)
+		_, _, _, _, err = q.DequeueByID(context.Background(), one, uuid.Nil)
 		require.Equal(t, jobqueue.ErrNotPending, err)
 	})
 }
@@ -571,7 +572,7 @@ func testMultipleChannels(t *testing.T, q jobqueue.JobQueue) {
 		go func() {
 			defer wg.Done()
 
-			id, _, _, _, _, err := q.Dequeue(context.Background(), []string{"octopus"}, []string{"kingfisher"})
+			id, _, _, _, _, err := q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{"kingfisher"})
 			require.NoError(t, err)
 
 			expectedID := <-twoChan
@@ -583,7 +584,7 @@ func testMultipleChannels(t *testing.T, q jobqueue.JobQueue) {
 		go func() {
 			defer wg.Done()
 
-			id, _, _, _, _, err := q.Dequeue(context.Background(), []string{"octopus"}, []string{"toucan"})
+			id, _, _, _, _, err := q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{"toucan"})
 			require.NoError(t, err)
 
 			expectedID := <-oneChan
@@ -610,13 +611,13 @@ func testMultipleChannels(t *testing.T, q jobqueue.JobQueue) {
 		go func() {
 			defer wg.Done()
 
-			id, _, _, _, _, err := q.Dequeue(context.Background(), []string{"octopus"}, []string{"kingfisher", "toucan"})
+			id, _, _, _, _, err := q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{"kingfisher", "toucan"})
 			require.NoError(t, err)
 
 			expectedID := <-oneChan
 			require.Equal(t, expectedID, id)
 
-			id, _, _, _, _, err = q.Dequeue(context.Background(), []string{"octopus"}, []string{"kingfisher", "toucan"})
+			id, _, _, _, _, err = q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{"kingfisher", "toucan"})
 			require.NoError(t, err)
 
 			expectedID = <-twoChan
@@ -639,11 +640,11 @@ func testMultipleChannels(t *testing.T, q jobqueue.JobQueue) {
 		// dequeue from an empty channel
 		ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Millisecond*100))
 		defer cancel()
-		_, _, _, _, _, err := q.Dequeue(ctx, []string{"octopus"}, []string{""})
+		_, _, _, _, _, err := q.Dequeue(ctx, uuid.Nil, []string{"octopus"}, []string{""})
 		require.ErrorIs(t, err, jobqueue.ErrDequeueTimeout)
 
 		// dequeue from toucan channel
-		_, _, _, _, _, err = q.Dequeue(context.Background(), []string{"octopus"}, []string{"toucan"})
+		_, _, _, _, _, err = q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{"toucan"})
 		require.NoError(t, err)
 
 		// enqueue into an empty channel
@@ -652,11 +653,11 @@ func testMultipleChannels(t *testing.T, q jobqueue.JobQueue) {
 		// dequeue from toucan channel
 		ctx2, cancel2 := context.WithDeadline(context.Background(), time.Now().Add(time.Millisecond*100))
 		defer cancel2()
-		_, _, _, _, _, err = q.Dequeue(ctx2, []string{"octopus"}, []string{"toucan"})
+		_, _, _, _, _, err = q.Dequeue(ctx2, uuid.Nil, []string{"octopus"}, []string{"toucan"})
 		require.ErrorIs(t, err, jobqueue.ErrDequeueTimeout)
 
 		// dequeue from an empty channel
-		_, _, _, _, _, err = q.Dequeue(context.Background(), []string{"octopus"}, []string{""})
+		_, _, _, _, _, err = q.Dequeue(context.Background(), uuid.Nil, []string{"octopus"}, []string{""})
 		require.NoError(t, err)
 	})
 }
@@ -700,3 +701,42 @@ func test100dequeuers(t *testing.T, q jobqueue.JobQueue) {
 	wg.Wait()
 
 }
+
+// Registers workers and runs jobs against them
+func testWorkers(t *testing.T, q jobqueue.JobQueue) {
+	one := pushTestJob(t, q, "octopus", nil, nil, "")
+
+	w1, err := q.InsertWorker("x86_64")
+	require.NoError(t, err)
+	w2, err := q.InsertWorker("aarch64")
+	require.NoError(t, err)
+
+	workers, err := q.Workers(0)
+	require.NoError(t, err)
+	require.Len(t, workers, 2)
+
+	workers, err = q.Workers(time.Hour * 24)
+	require.NoError(t, err)
+	require.Len(t, workers, 0)
+
+	_, _, _, _, _, err = q.Dequeue(context.Background(), w1, []string{"octopus"}, []string{""})
+	require.NoError(t, err)
+
+	err = q.DeleteWorker(w1)
+	require.Equal(t, err, jobqueue.ErrActiveJobs)
+
+	err = q.UpdateWorkerStatus(w1)
+	require.NoError(t, err)
+
+	err = q.UpdateWorkerStatus(uuid.New())
+	require.Equal(t, err, jobqueue.ErrWorkerNotExist)
+
+	err = q.RequeueOrFinishJob(one, 0, &testResult{})
+	require.NoError(t, err)
+
+	err = q.DeleteWorker(w1)
+	require.NoError(t, err)
+
+	err = q.DeleteWorker(w2)
+	require.NoError(t, err)
+}
diff --git a/internal/test/helpers.go b/internal/test/helpers.go
index a34dec347..1b845c586 100644
--- a/internal/test/helpers.go
+++ b/internal/test/helpers.go
@@ -114,6 +114,7 @@ func TestRouteWithReply(t *testing.T, api http.Handler, external bool, method, p
 
 	if expectedJSON == "" {
 		require.Lenf(t, replyJSON, 0, "%s: expected no response body, but got:\n%s", path, replyJSON)
+		return
 	}
 
 	if expectedJSON == "?" {
diff --git a/internal/weldr/compose_test.go b/internal/weldr/compose_test.go
index 628c6956f..38907c270 100644
--- a/internal/weldr/compose_test.go
+++ b/internal/weldr/compose_test.go
@@ -6,6 +6,7 @@ import (
 	"os"
 	"testing"
 
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/require"
 
 	"github.com/osbuild/images/pkg/distro"
@@ -44,7 +45,7 @@ func TestComposeStatusFromLegacyError(t *testing.T) {
 	jobId, err := api.workers.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: mf}, "")
 	require.NoError(t, err)
 
-	j, token, _, _, _, err := api.workers.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""})
+	j, token, _, _, _, err := api.workers.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, jobId, j)
 
@@ -97,7 +98,7 @@ func TestComposeStatusFromJobError(t *testing.T) {
 	jobId, err := api.workers.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: mf}, "")
 	require.NoError(t, err)
 
-	j, token, _, _, _, err := api.workers.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""})
+	j, token, _, _, _, err := api.workers.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, jobId, j)
 
diff --git a/internal/worker/client_test.go b/internal/worker/client_test.go
index 93a9b6b8f..78268e69d 100644
--- a/internal/worker/client_test.go
+++ b/internal/worker/client_test.go
@@ -128,11 +128,12 @@ func TestProxy(t *testing.T) {
 	require.False(t, c)
 	require.NoError(t, err)
 
-	// we expect 5 calls to go through the proxy:
+	// we expect 6 calls to go through the proxy:
+	// - register worker
 	// - request job (fails, no oauth token)
 	// - oauth call
 	// - request job (succeeds)
 	// - upload artifact
 	// - cancel
-	require.Equal(t, 5, proxy.calls)
+	require.Equal(t, 6, proxy.calls)
 }
diff --git a/internal/worker/server_test.go b/internal/worker/server_test.go
index eaa7fbd81..d90faef0a 100644
--- a/internal/worker/server_test.go
+++ b/internal/worker/server_test.go
@@ -22,24 +22,21 @@ import (
 	"github.com/osbuild/images/pkg/osbuild"
 	"github.com/osbuild/images/pkg/platform"
 	"github.com/osbuild/images/pkg/rpmmd"
+	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/jobqueue/fsjobqueue"
 	"github.com/osbuild/osbuild-composer/internal/target"
 	"github.com/osbuild/osbuild-composer/internal/test"
 	"github.com/osbuild/osbuild-composer/internal/worker"
+	"github.com/osbuild/osbuild-composer/internal/worker/api"
 	"github.com/osbuild/osbuild-composer/internal/worker/clienterrors"
 )
 
-func newTestServer(t *testing.T, tempdir string, jobRequestTimeout time.Duration, basePath string, acceptArtifacts bool) *worker.Server {
+func newTestServer(t *testing.T, tempdir string, config worker.Config, acceptArtifacts bool) *worker.Server {
 	q, err := fsjobqueue.New(tempdir)
 	if err != nil {
 		t.Fatalf("error creating fsjobqueue: %v", err)
 	}
 
-	config := worker.Config{
-		RequestJobTimeout: jobRequestTimeout,
-		BasePath:          basePath,
-	}
-
 	if acceptArtifacts {
 		artifactsDir := path.Join(tempdir, "artifacts")
 		err := os.Mkdir(artifactsDir, 0755)
@@ -52,9 +49,14 @@ func newTestServer(t *testing.T, tempdir string, jobRequestTimeout time.Duration
 	return worker.NewServer(nil, q, config)
 }
 
+var defaultConfig = worker.Config{
+	RequestJobTimeout: time.Duration(0),
+	BasePath:          "/api/worker/v1",
+}
+
 // Ensure that the status request returns OK.
 func TestStatus(t *testing.T) {
-	server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/worker/v1", false)
+	server := newTestServer(t, t.TempDir(), defaultConfig, false)
 	handler := server.Handler()
 	test.TestRoute(t, handler, false, "GET", "/api/worker/v1/status", ``, http.StatusOK, `{"status":"OK", "href": "/api/worker/v1/status", "kind":"Status"}`, "message", "id")
 }
@@ -84,7 +86,7 @@ func TestErrors(t *testing.T) {
 	tempdir := t.TempDir()
 
 	for _, c := range cases {
-		server := newTestServer(t, tempdir, time.Duration(0), "/api/worker/v1", false)
+		server := newTestServer(t, tempdir, defaultConfig, false)
 		handler := server.Handler()
 		test.TestRoute(t, handler, false, c.Method, c.Path, c.Body, c.ExpectedStatus, `{"kind":"Error"}`, "message", "href", "operation_id", "reason", "id", "code")
 	}
@@ -113,9 +115,13 @@ func TestErrorsAlteredBasePath(t *testing.T) {
 	}
 
 	tempdir := t.TempDir()
+	config := worker.Config{
+		RequestJobTimeout: time.Duration(0),
+		BasePath:          "/api/image-builder-worker/v1",
+	}
 
 	for _, c := range cases {
-		server := newTestServer(t, tempdir, time.Duration(0), "/api/image-builder-worker/v1", false)
+		server := newTestServer(t, tempdir, config, false)
 		handler := server.Handler()
 		test.TestRoute(t, handler, false, c.Method, c.Path, c.Body, c.ExpectedStatus, `{"kind":"Error"}`, "message", "href", "operation_id", "reason", "id", "code")
 	}
@@ -139,7 +145,7 @@ func TestCreate(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
 	}
-	server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/worker/v1", false)
+	server := newTestServer(t, t.TempDir(), defaultConfig, false)
 	handler := server.Handler()
 
 	_, err = server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: mf}, "")
@@ -165,7 +171,7 @@ func TestCancel(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
 	}
-	server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/worker/v1", false)
+	server := newTestServer(t, t.TempDir(), defaultConfig, false)
 	mf, err := manifest.Serialize(nil, nil, nil)
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
@@ -175,7 +181,7 @@ func TestCancel(t *testing.T) {
 	jobId, err := server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: mf}, "")
 	require.NoError(t, err)
 
-	j, token, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""})
+	j, token, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, jobId, j)
 	require.Equal(t, worker.JobTypeOSBuild, typ)
@@ -206,7 +212,7 @@ func TestUpdate(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
 	}
-	server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/worker/v1", false)
+	server := newTestServer(t, t.TempDir(), defaultConfig, false)
 	mf, err := manifest.Serialize(nil, nil, nil)
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
@@ -216,7 +222,7 @@ func TestUpdate(t *testing.T) {
 	jobId, err := server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: mf}, "")
 	require.NoError(t, err)
 
-	j, token, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""})
+	j, token, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, jobId, j)
 	require.Equal(t, worker.JobTypeOSBuild, typ)
@@ -261,11 +267,11 @@ func TestArgs(t *testing.T) {
 		},
 	}
 
-	server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/worker/v1", false)
+	server := newTestServer(t, t.TempDir(), defaultConfig, false)
 	jobId, err := server.EnqueueOSBuild(arch.Name(), &job, "")
 	require.NoError(t, err)
 
-	_, _, _, args, _, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""})
+	_, _, _, args, _, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.NotNil(t, args)
 
@@ -289,7 +295,7 @@ func TestUpload(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
 	}
-	server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/worker/v1", true)
+	server := newTestServer(t, t.TempDir(), defaultConfig, true)
 	mf, err := manifest.Serialize(nil, nil, nil)
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
@@ -299,7 +305,7 @@ func TestUpload(t *testing.T) {
 	jobID, err := server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: mf}, "")
 	require.NoError(t, err)
 
-	j, token, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""})
+	j, token, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, jobID, j)
 	require.Equal(t, worker.JobTypeOSBuild, typ)
@@ -323,7 +329,7 @@ func TestUploadNotAcceptingArtifacts(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error creating osbuild manifest: %v", err)
 	}
-	server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/worker/v1", false)
+	server := newTestServer(t, t.TempDir(), defaultConfig, false)
 	handler := server.Handler()
 	mf, _ := manifest.Serialize(nil, nil, nil)
 	if err != nil {
@@ -333,7 +339,7 @@ func TestUploadNotAcceptingArtifacts(t *testing.T) {
 	jobID, err := server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: mf}, "")
 	require.NoError(t, err)
 
-	j, token, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""})
+	j, token, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, jobID, j)
 	require.Equal(t, worker.JobTypeOSBuild, typ)
@@ -362,13 +368,18 @@ func TestUploadAlteredBasePath(t *testing.T) {
 		t.Fatalf("error creating osbuild manifest: %v", err)
 	}
 
-	server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/image-builder-worker/v1", true)
+	config := worker.Config{
+		RequestJobTimeout: time.Duration(0),
+		BasePath:          "/api/image-builder-worker/v1",
+	}
+
+	server := newTestServer(t, t.TempDir(), config, true)
 	handler := server.Handler()
 
 	jobID, err := server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: mf}, "")
 	require.NoError(t, err)
 
-	j, token, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""})
+	j, token, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 	require.Equal(t, jobID, j)
 	require.Equal(t, worker.JobTypeOSBuild, typ)
@@ -384,9 +395,13 @@ func TestTimeout(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error getting arch from distro: %v", err)
 	}
-	server := newTestServer(t, t.TempDir(), time.Millisecond*10, "/api/image-builder-worker/v1", false)
+	config := worker.Config{
+		RequestJobTimeout: time.Millisecond * 10,
+		BasePath:          "/api/image-builder-worker/v1",
+	}
+	server := newTestServer(t, t.TempDir(), config, false)
 
-	_, _, _, _, _, err = server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""})
+	_, _, _, _, _, err = server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 	require.Equal(t, jobqueue.ErrDequeueTimeout, err)
 
 	test.TestRoute(t, server.Handler(), false, "POST", "/api/image-builder-worker/v1/jobs", `{"arch":"arch","types":["types"]}`, http.StatusNoContent,
@@ -399,7 +414,7 @@ func TestRequestJobById(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error getting arch from distro: %v", err)
 	}
-	server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/worker/v1", false)
+	server := newTestServer(t, t.TempDir(), defaultConfig, false)
 	handler := server.Handler()
 
 	depsolveJobId, err := server.EnqueueDepsolve(&worker.DepsolveJob{}, "")
@@ -414,7 +429,7 @@ func TestRequestJobById(t *testing.T) {
 	_, _, _, _, _, err = server.RequestJobById(context.Background(), arch.Name(), jobId)
 	require.Error(t, jobqueue.ErrNotPending, err)
 
-	_, token, _, _, _, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeDepsolve}, []string{""})
+	_, token, _, _, _, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeDepsolve}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 
 	depsolveJR, err := json.Marshal(worker.DepsolveJobResult{})
@@ -440,7 +455,11 @@ func TestMixedOSBuildJob(t *testing.T) {
 	require := require.New(t)
 
 	emptyManifestV2 := manifest.OSBuildManifest(`{"version":"2","pipelines":{}}`)
-	server := newTestServer(t, t.TempDir(), time.Millisecond*10, "/", false)
+	config := worker.Config{
+		RequestJobTimeout: time.Millisecond * 10,
+		BasePath:          "/",
+	}
+	server := newTestServer(t, t.TempDir(), config, false)
 	fbPipelines := &worker.PipelineNames{Build: distro.BuildPipelinesFallback(), Payload: distro.PayloadPipelinesFallback()}
 
 	oldJob := worker.OSBuildJob{
@@ -506,7 +525,7 @@ func TestMixedOSBuildJob(t *testing.T) {
 		// don't block forever if the jobs weren't added or can't be retrieved
 		ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
 		defer cancel()
-		id, token, _, _, _, err := server.RequestJob(ctx, "x", []string{worker.JobTypeOSBuild}, []string{""})
+		id, token, _, _, _, err := server.RequestJob(ctx, "x", []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
 		require.NoError(err)
 		return id, token
 	}
@@ -598,12 +617,12 @@ func TestDepsolveLegacyErrorConversion(t *testing.T) {
 	if err != nil {
 		t.Fatalf("error getting arch from distro: %v", err)
 	}
-	server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/worker/v1", false)
+	server := newTestServer(t, t.TempDir(), defaultConfig, false)
 
 	depsolveJobId, err := server.EnqueueDepsolve(&worker.DepsolveJob{}, "")
 	require.NoError(t, err)
 
-	_, _, _, _, _, err = server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeDepsolve}, []string{""})
+	_, _, _, _, _, err = server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeDepsolve}, []string{""}, uuid.Nil)
 	require.NoError(t, err)
 
 	reason := "Depsolve failed"
@@ -1579,7 +1598,7 @@ func TestJobDependencyChainErrors(t *testing.T) {
 
 	for idx, c := range cases {
 		t.Logf("Test case #%d", idx)
-		server := newTestServer(t, t.TempDir(), time.Duration(0), "/api/worker/v1", false)
+		server := newTestServer(t, t.TempDir(), defaultConfig, false)
 		ids, err := enqueueAndFinishTestJobDependencies(server, []testJob{c.job})
 		require.Nil(t, err)
 		require.Len(t, ids, 1)
@@ -1590,3 +1609,64 @@ func TestJobDependencyChainErrors(t *testing.T) {
 		assert.EqualValues(t, c.expectedError, errors)
 	}
 }
+
+func TestWorkerWatch(t *testing.T) {
+	config := worker.Config{
+		RequestJobTimeout: time.Duration(0),
+		BasePath:          "/api/worker/v1",
+		WorkerTimeout:     time.Millisecond * 200,
+		WorkerWatchFreq:   time.Millisecond * 100,
+	}
+
+	server := newTestServer(t, t.TempDir(), config, false)
+
+	reply := test.TestRouteWithReply(t, server.Handler(), false, "POST", "/api/worker/v1/workers", fmt.Sprintf(`{"arch":"%s"}`, common.CurrentArch()), 201, `{"href":"/api/worker/v1/workers","kind":"WorkerID","id": "15"}`, "id", "worker_id")
+	var resp api.PostWorkersResponse
+	require.NoError(t, json.Unmarshal(reply, &resp))
+	workerID, err := uuid.Parse(resp.WorkerId)
+	require.NoError(t, err)
+
+	test.TestRoute(t, server.Handler(), false, "POST", fmt.Sprintf("/api/worker/v1/workers/%s/status", workerID), "{}", 200, "")
+	time.Sleep(time.Millisecond * 400)
+	test.TestRoute(t, server.Handler(), false, "POST", fmt.Sprintf("/api/worker/v1/workers/%s/status", workerID), "", 400,
+		`{"href":"/api/worker/v1/errors/18","code":"IMAGE-BUILDER-WORKER-18","id":"18","kind":"Error","message":"Given worker id doesn't exist","reason":"Given worker id doesn't exist"}`,
+		"operation_id")
+}
+
+func TestRequestJobForWorker(t *testing.T) {
+	server := newTestServer(t, t.TempDir(), defaultConfig, false)
+	reply := test.TestRouteWithReply(t, server.Handler(), false, "POST", "/api/worker/v1/workers", fmt.Sprintf(`{"arch":"%s"}`, common.CurrentArch()), 201, `{"href":"/api/worker/v1/workers","kind":"WorkerID","id": "15"}`, "id", "worker_id")
+	var resp api.PostWorkersResponse
+	require.NoError(t, json.Unmarshal(reply, &resp))
+	workerID, err := uuid.Parse(resp.WorkerId)
+	require.NoError(t, err)
+	test.TestRoute(t, server.Handler(), false, "POST", fmt.Sprintf("/api/worker/v1/workers/%s/status", workerID), "{}", 200, "")
+
+	distroStruct := test_distro.New()
+	arch, err := distroStruct.GetArch(test_distro.TestArchName)
+	if err != nil {
+		t.Fatalf("error getting arch from distro: %v", err)
+	}
+	imageType, err := arch.GetImageType(test_distro.TestImageTypeName)
+	if err != nil {
+		t.Fatalf("error getting image type from arch: %v", err)
+	}
+	manifest, _, err := imageType.Manifest(nil, distro.ImageOptions{Size: imageType.Size(0)}, nil, 0)
+	if err != nil {
+		t.Fatalf("error creating osbuild manifest: %v", err)
+	}
+	mf, err := manifest.Serialize(nil, nil, nil)
+	if err != nil {
+		t.Fatalf("error creating osbuild manifest: %v", err)
+	}
+	jobId, err := server.EnqueueOSBuild(arch.Name(), &worker.OSBuildJob{Manifest: mf}, "")
+	require.NoError(t, err)
+
+	// Can request a job with worker ID
+	j, _, typ, args, dynamicArgs, err := server.RequestJob(context.Background(), arch.Name(), []string{worker.JobTypeOSBuild}, []string{""}, workerID)
+	require.NoError(t, err)
+	require.Equal(t, jobId, j)
+	require.Equal(t, worker.JobTypeOSBuild, typ)
+	require.NotNil(t, args)
+	require.Nil(t, dynamicArgs)
+}