jobqueue: Add DeleteJob function

This allows jobs to be deleted from the database.
Currently only implemented by fsjobqueue. The function for
dbjobqueue currently returns nil.

This will remove all the job files used by the root job UUID as long as
no other job depends on them. ie. It starts at the top, and moves down
the dependency tree until it finds a job that is also used by another
job, removes the job to be deleted from its dependants list, and moves
back up the tree only deleting jobs with empty dependants lists.

Related: RHEL-60120

internal/jobqueue/fsjobqueue/fsjobqueue.go

@@ -726,3 +726,48 @@ func (q *fsJobQueue) AllRootJobIDs() ([]uuid.UUID, error) {
 
 	return jobIDs, nil
 }
+
+// DeleteJob will delete a job and all of its dependencies
+// If a dependency has multiple depenents it will only delete the parent job from
+// the dependants list and then re-save the job instead of removing it.
+//
+// This assumes that the jobs have been created correctly, and that they have
+// no dependency loops. Shared Dependants are ok, but a job cannot have a dependancy
+// on any of its parents (this should never happen).
+func (q *fsJobQueue) DeleteJob(_ context.Context, id uuid.UUID) error {
+	// Start it off with an empty parent
+	return q.deleteJob(uuid.UUID{}, id)
+}
+
+// deleteJob will delete jobs as far down the list as possible
+// missing dependencies are ignored, it deletes as much as it can.
+// A missing parent (the first call) will be returned as an error
+func (q *fsJobQueue) deleteJob(parent, id uuid.UUID) error {
+	var j job
+	_, err := q.db.Read(id.String(), &j)
+	if err != nil {
+		return err
+	}
+
+	// Delete the parent uuid from the Dependents list
+	var deps []uuid.UUID
+	for _, d := range j.Dependents {
+		if d == parent {
+			continue
+		}
+		deps = append(deps, d)
+	}
+	j.Dependents = deps
+
+	// This job can only be deleted when the Dependents list is empty
+	// Otherwise it needs to be saved with the new Dependents list
+	if len(j.Dependents) > 0 {
+		return q.db.Write(id.String(), j)
+	}
+	// Recursively delete the dependencies of this job
+	for _, dj := range j.Dependencies {
+		_ = q.deleteJob(id, dj)
+	}
+
+	return q.db.Delete(id.String())
+}

internal/jobqueue/fsjobqueue/fsjobqueue_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
 	"github.com/osbuild/osbuild-composer/internal/jobqueue/fsjobqueue"
@@ -87,3 +88,82 @@ func TestAllRootJobIDs(t *testing.T) {
 	sortUUIDs(roots)
 	require.Equal(t, rootJobs, roots)
 }
+
+func TestDeleteJob(t *testing.T) {
+	dir := t.TempDir()
+	q, err := fsjobqueue.New(dir)
+	require.Nil(t, err)
+	require.NotNil(t, q)
+
+	// root with no dependencies
+	jidRoot1, err := q.Enqueue("oneRoot", nil, nil, "OneRootJob")
+	require.Nil(t, err)
+
+	err = q.DeleteJob(context.TODO(), jidRoot1)
+	require.Nil(t, err)
+	jobs, err := q.AllRootJobIDs()
+	require.Nil(t, err)
+	require.Equal(t, 0, len(jobs))
+
+	// root with 2 dependencies
+	jid1, err := q.Enqueue("twoDeps", nil, nil, "TwoDepJobs")
+	require.Nil(t, err)
+	jid2, err := q.Enqueue("twoDeps", nil, nil, "TwoDepJobs")
+	require.Nil(t, err)
+	jidRoot2, err := q.Enqueue("twoDeps", nil, []uuid.UUID{jid1, jid2}, "TwoDepJobs")
+	require.Nil(t, err)
+
+	// root with 2 dependencies, one shared with the previous root
+	jid3, err := q.Enqueue("sharedDeps", nil, nil, "SharedDepJobs")
+	require.Nil(t, err)
+	jidRoot3, err := q.Enqueue("sharedDeps", nil, []uuid.UUID{jid1, jid3}, "SharedDepJobs")
+	require.Nil(t, err)
+
+	// This should only remove jidRoot2 and jid2, leaving jidRoot3, jid1, jid3
+	err = q.DeleteJob(context.TODO(), jidRoot2)
+	require.Nil(t, err)
+	jobs, err = q.AllRootJobIDs()
+	require.Nil(t, err)
+	require.Equal(t, 1, len(jobs))
+	assert.Equal(t, []uuid.UUID{jidRoot3}, jobs)
+
+	// This should remove the rest
+	err = q.DeleteJob(context.TODO(), jidRoot3)
+	require.Nil(t, err)
+	jobs, err = q.AllRootJobIDs()
+	require.Nil(t, err)
+	require.Equal(t, 0, len(jobs))
+
+	// Make sure all the jobs are deleted
+	allJobs := []uuid.UUID{jidRoot1, jidRoot2, jidRoot3, jid1, jid2, jid3}
+	for _, jobId := range allJobs {
+		jobType, _, _, _, err := q.Job(jobId)
+		assert.Error(t, err, jobType)
+	}
+
+	// root with 2 jobs depending on another (simulates Koji jobs)
+	kojiOSTree, err := q.Enqueue("ostree", nil, nil, "KojiJob")
+	require.Nil(t, err)
+	kojiDepsolve, err := q.Enqueue("depsolve", nil, nil, "KojiJob")
+	require.Nil(t, err)
+	kojiManifest, err := q.Enqueue("manifest", nil, []uuid.UUID{kojiOSTree, kojiDepsolve}, "KojiJob")
+	require.Nil(t, err)
+	kojiInit, err := q.Enqueue("init", nil, nil, "KojiJob")
+	require.Nil(t, err)
+	kojiRoot, err := q.Enqueue("final", nil, []uuid.UUID{kojiInit, kojiManifest, kojiDepsolve}, "KojiJob")
+	require.Nil(t, err)
+
+	// Delete the koji job
+	err = q.DeleteJob(context.TODO(), kojiRoot)
+	require.Nil(t, err)
+	jobs, err = q.AllRootJobIDs()
+	require.Nil(t, err)
+	require.Equal(t, 0, len(jobs))
+
+	// Make sure all the jobs are deleted
+	kojiJobs := []uuid.UUID{kojiRoot, kojiInit, kojiOSTree, kojiDepsolve, kojiManifest}
+	for _, jobId := range kojiJobs {
+		jobType, _, _, _, err := q.Job(jobId)
+		assert.Error(t, err, jobType)
+	}
+}

pkg/jobqueue/dbjobqueue/dbjobqueue.go

@@ -910,3 +910,9 @@ func (q *DBJobQueue) AllRootJobIDs() ([]uuid.UUID, error) {
 
 	return nil, nil
 }
+
+// DeleteJob deletes a job from the database
+// Currently not implemented for the DBJobQueue
+func (q *DBJobQueue) DeleteJob(ctx context.Context, id uuid.UUID) error {
+	return nil
+}

pkg/jobqueue/jobqueue.go

@@ -98,6 +98,9 @@ type JobQueue interface {
 
 	// AllRootJobIDs returns a list of top level job UUIDs that the worker knows about
 	AllRootJobIDs() ([]uuid.UUID, error)
+
+	// DeleteJob deletes a job and all of its dependencies
+	DeleteJob(context.Context, uuid.UUID) error
 }
 
 // SimpleLogger provides a structured logging methods for the jobqueue library.