update to panic instead of fatal, add more tests, fix job tests

Author: siherrmann

database/dbJob.go

@@ -97,7 +97,7 @@ func (r JobDBHandler) CreateTable() error {
 		SELECT create_hypertable('job_archive', by_range('updated_at'), if_not_exists => TRUE);`,
 	)
 	if err != nil {
-		log.Fatalf("error creating job table: %#v", err)
+		log.Panicf("error creating job table: %#v", err)
 	}
 
 	_, err = r.db.Instance.ExecContext(
@@ -107,7 +107,7 @@ func (r JobDBHandler) CreateTable() error {
 			FOR EACH ROW EXECUTE PROCEDURE notify_event();`,
 	)
 	if err != nil {
-		log.Fatalf("error creating notify trigger on job table: %#v", err)
+		log.Panicf("error creating notify trigger on job table: %#v", err)
 	}
 
 	err = r.db.CreateIndexes("job", "worker_id", "worker_rid", "status", "created_at", "updated_at") // Indexes on common search/filter fields

database/dbWorker.go

@@ -78,7 +78,7 @@ func (r WorkerDBHandler) CreateTable() error {
 		)`,
 	)
 	if err != nil {
-		log.Fatalf("error creating worker table: %#v", err)
+		log.Panicf("error creating worker table: %#v", err)
 	}
 
 	err = r.db.CreateIndexes("worker", "rid", "name", "status")

helper/database.go

@@ -34,7 +34,7 @@ func NewDatabase(name string, dbConfig *DatabaseConfiguration) *Database {
 
 		err := db.AddNotifyFunction()
 		if err != nil {
-			logger.Fatalf("failed to add notify function: %v", err)
+			logger.Panicf("failed to add notify function: %v", err)
 		}
 
 		return db
@@ -87,7 +87,7 @@ func (d *Database) ConnectToDatabase(dbConfig *DatabaseConfiguration, logger *lo
 	connectOnce.Do(func() {
 		db, err = sql.Open("postgres", dbConfig.DatabaseConnectionString())
 		if err != nil {
-			logger.Fatalf("error establishing connection to db: %v. Trying again.", err.Error())
+			logger.Panicf("error establishing connection to db: %v. Trying again.", err.Error())
 		}
 
 		db.SetMaxOpenConns(10)
@@ -307,7 +307,7 @@ func (d *Database) Health() map[string]string {
 	if err != nil {
 		stats["status"] = "down"
 		stats["error"] = fmt.Sprintf("db down: %v", err)
-		log.Fatalf("db down: %v", err) // Log the error and terminate the program
+		log.Panicf("db down: %v", err) // Log the error and terminate the program
 		return stats
 	}
 

main_test.go

@@ -13,6 +13,8 @@ import (
 	"github.com/testcontainers/testcontainers-go"
 )
 
+const maxDeviation = 50 * time.Millisecond
+
 var dbPort string
 
 func TestMain(m *testing.M) {

queuer.go

@@ -35,14 +35,19 @@ type Queuer struct {
 	log *log.Logger
 }
 
+// NewQueuer creates a new Queuer instance with the given name and max concurrency.
+// It initializes the database connection, job listeners, and worker.
+// If options are provided, it creates a worker with those options.
+// If any error occurs during initialization, it logs a panic error and exits the program.
+// It returns a pointer to the newly created Queuer instance.
 func NewQueuer(name string, maxConcurrency int, options ...*model.OnError) *Queuer {
 	// Logger
 	logger := log.New(os.Stdout, "Queuer: ", log.Ltime)
 
 	// Database
 	dbConfig, err := helper.NewDatabaseConfiguration()
 	if err != nil {
-		logger.Fatalf("failed to create database configuration: %v", err)
+		logger.Panicf("failed to create database configuration: %v", err)
 	}
 	dbConnection := helper.NewDatabase(
 		"queuer",
@@ -54,44 +59,44 @@ func NewQueuer(name string, maxConcurrency int, options ...*model.OnError) *Queu
 	var dbWorker database.WorkerDBHandlerFunctions
 	dbJob, err = database.NewJobDBHandler(dbConnection)
 	if err != nil {
-		logger.Fatalf("failed to create job db handler: %v", err)
+		logger.Panicf("failed to create job db handler: %v", err)
 	}
 	dbWorker, err = database.NewWorkerDBHandler(dbConnection)
 	if err != nil {
-		logger.Fatalf("failed to create worker db handler: %v", err)
+		logger.Panicf("failed to create worker db handler: %v", err)
 	}
 
 	// Job listeners
 	jobInsertListener, err := database.NewQueuerListener(dbConfig, "job.INSERT")
 	if err != nil {
-		logger.Fatalf("failed to create job insert listener: %v", err)
+		logger.Panicf("failed to create job insert listener: %v", err)
 	}
 	jobUpdateListener, err := database.NewQueuerListener(dbConfig, "job.UPDATE")
 	if err != nil {
-		logger.Fatalf("failed to create job update listener: %v", err)
+		logger.Panicf("failed to create job update listener: %v", err)
 	}
 	jobDeleteListener, err := database.NewQueuerListener(dbConfig, "job.DELETE")
 	if err != nil {
-		logger.Fatalf("failed to create job delete listener: %v", err)
+		logger.Panicf("failed to create job delete listener: %v", err)
 	}
 
 	// Inserting worker
 	var newWorker *model.Worker
 	if len(options) > 0 {
 		newWorker, err = model.NewWorkerWithOptions(name, maxConcurrency, options[0])
 		if err != nil {
-			logger.Fatalf("error creating new worker with options: %v", err)
+			logger.Panicf("error creating new worker with options: %v", err)
 		}
 	} else {
 		newWorker, err = model.NewWorker(name, maxConcurrency)
 		if err != nil {
-			logger.Fatalf("error creating new worker: %v", err)
+			logger.Panicf("error creating new worker: %v", err)
 		}
 	}
 
 	worker, err := dbWorker.InsertWorker(newWorker)
 	if err != nil {
-		logger.Fatalf("error inserting worker: %v", err)
+		logger.Panicf("error inserting worker: %v", err)
 	}
 	logger.Printf("Worker %s created with RID %s", worker.Name, worker.RID.String())
 
@@ -108,9 +113,12 @@ func NewQueuer(name string, maxConcurrency int, options ...*model.OnError) *Queu
 	}
 }
 
+// Start starts the queuer by initializing the job listeners and starting the job poll ticker.
+// It checks if the queuer is initialized properly, and if not, it logs a panic error and exits the program.
+// It runs the job processing in a separate goroutine and listens for job events.
 func (q *Queuer) Start(ctx context.Context, cancel context.CancelFunc) {
 	if q.dbJob == nil || q.dbWorker == nil || q.jobInsertListener == nil || q.jobUpdateListener == nil || q.jobDeleteListener == nil {
-		q.log.Fatalln("worker is not initialized properly")
+		q.log.Panicln("worker is not initialized properly")
 	}
 
 	q.ctx = ctx
@@ -135,6 +143,8 @@ func (q *Queuer) Start(ctx context.Context, cancel context.CancelFunc) {
 	}()
 }
 
+// Stop stops the queuer by closing the job listeners, cancelling all queued and running jobs,
+// and cancelling the context to stop the queuer.
 func (q *Queuer) Stop() error {
 	if q.jobInsertListener != nil {
 		err := q.jobInsertListener.Listener.Close()
@@ -156,7 +166,7 @@ func (q *Queuer) Stop() error {
 	}
 
 	// Cancel all queued and running jobs
-	err := q.CancelAllJobsByWorker(q.worker.RID)
+	err := q.CancelAllJobsByWorker(q.worker.RID, 100)
 	if err != nil {
 		return fmt.Errorf("error cancelling all jobs by worker: %v", err)
 	}

queuerJob.go

@@ -80,8 +80,8 @@ func (q *Queuer) CancelJob(jobRid uuid.UUID) (*model.Job, error) {
 	return job, nil
 }
 
-func (q *Queuer) CancelAllJobsByWorker(workerRid uuid.UUID) error {
-	jobs, err := q.dbJob.SelectAllJobsByWorkerRID(workerRid, 0, 0)
+func (q *Queuer) CancelAllJobsByWorker(workerRid uuid.UUID, entries int) error {
+	jobs, err := q.dbJob.SelectAllJobsByWorkerRID(workerRid, 0, entries)
 	if err != nil {
 		return fmt.Errorf("error selecting jobs by worker RID %v: %v", workerRid, err)
 	}
@@ -97,21 +97,21 @@ func (q *Queuer) CancelAllJobsByWorker(workerRid uuid.UUID) error {
 }
 
 // ReaddJobFromArchive readds a job from the archive back to the queue.
-func (q *Queuer) ReaddJobFromArchive(jobRid uuid.UUID) error {
+func (q *Queuer) ReaddJobFromArchive(jobRid uuid.UUID) (*model.Job, error) {
 	job, err := q.dbJob.SelectJobFromArchive(jobRid)
 	if err != nil {
-		return fmt.Errorf("error selecting job from archive with rid %v: %v", jobRid, err)
+		return nil, fmt.Errorf("error selecting job from archive with rid %v: %v", jobRid, err)
 	}
 
 	// Readd the job to the queue
 	newJob, err := q.AddJobWithOptions(job.Options, job.TaskName, job.Parameters...)
 	if err != nil {
-		return fmt.Errorf("error readding job: %v", err)
+		return nil, fmt.Errorf("error readding job: %v", err)
 	}
 
 	q.log.Printf("Job readded with RID %v", newJob.RID)
 
-	return nil
+	return newJob, nil
 }
 
 // GetJob retrieves a job by its RID.
@@ -285,8 +285,8 @@ func (q *Queuer) scheduleJob(job *model.Job) {
 }
 
 func (q *Queuer) cancelJob(job *model.Job) error {
-	var err error
-	if job.Status == model.JobStatusRunning {
+	switch job.Status {
+	case model.JobStatusRunning:
 		jobRunner, found := q.activeRunners.Load(job.RID)
 		if !found {
 			return fmt.Errorf("job with rid %v not found or not running", job.RID)
@@ -295,16 +295,16 @@ func (q *Queuer) cancelJob(job *model.Job) error {
 		runner := jobRunner.(*core.Runner)
 		runner.Cancel(func() {
 			job.Status = model.JobStatusCancelled
-			job, err = q.dbJob.UpdateJobFinal(job)
+			_, err := q.dbJob.UpdateJobFinal(job)
 			if err != nil {
 				q.log.Printf("error updating job status to cancelled: %v", err)
 			}
 		})
-	} else if job.Status == model.JobStatusScheduled || job.Status == model.JobStatusQueued {
+	case model.JobStatusScheduled, model.JobStatusQueued:
 		job.Status = model.JobStatusCancelled
-		job, err = q.dbJob.UpdateJobFinal(job)
+		_, err := q.dbJob.UpdateJobFinal(job)
 		if err != nil {
-			return fmt.Errorf("error updating job status to cancelled: %v", err)
+			q.log.Printf("error updating job status to cancelled: %v", err)
 		}
 	}
 	return nil