docs: improve and expand documentation for queue and worker internals

- Add comprehensive package-level and function-level documentation to clarify queue behavior, worker management, and shutdown procedures
- Replace brief or missing comments with detailed explanations for struct fields and internal logic
- Improve clarity of retry, scheduling, and worker lifecycle mechanisms through enhanced inline comments
- Refactor comments to use Go-style doc comments and block comments for exported types and functions

Signed-off-by: Bo-Yi Wu <[email protected]>

Author: appleboy

queue.go

@@ -1,5 +1,8 @@
 package queue
 
+// Package queue provides a high-performance, extensible message queue implementation
+// supporting multiple workers, job retries, dynamic scaling, and graceful shutdown.
+
 import (
 	"context"
 	"errors"
@@ -13,45 +16,52 @@ import (
 	"github.com/jpillora/backoff"
 )
 
-// ErrQueueShutdown the queue is released and closed.
+/*
+ErrQueueShutdown is returned when an operation is attempted on a queue
+that has already been closed and released.
+*/
 var ErrQueueShutdown = errors.New("queue has been closed and released")
 
 type (
-	// A Queue is a message queue.
+	// Queue represents a message queue with worker management, job scheduling,
+	// retry logic, and graceful shutdown capabilities.
 	Queue struct {
-		sync.Mutex
-		metric        *metric
-		logger        Logger
-		workerCount   int64
-		routineGroup  *routineGroup
-		quit          chan struct{}
-		ready         chan struct{}
-		notify        chan struct{}
-		worker        core.Worker
-		stopOnce      sync.Once
-		stopFlag      int32
-		afterFn       func()
-		retryInterval time.Duration
+		sync.Mutex                  // Mutex to protect concurrent access to queue state
+		metric        *metric       // Metrics collector for tracking queue and worker stats
+		logger        Logger        // Logger for queue events and errors
+		workerCount   int64         // Number of worker goroutines to process jobs
+		routineGroup  *routineGroup // Group to manage and wait for goroutines
+		quit          chan struct{} // Channel to signal shutdown to all goroutines
+		ready         chan struct{} // Channel to signal worker readiness
+		notify        chan struct{} // Channel to notify workers of new jobs
+		worker        core.Worker   // The worker implementation that processes jobs
+		stopOnce      sync.Once     // Ensures shutdown is only performed once
+		stopFlag      int32         // Atomic flag indicating if shutdown has started
+		afterFn       func()        // Optional callback after each job execution
+		retryInterval time.Duration // Interval for retrying job requests
 	}
 )
 
-// ErrMissingWorker missing define worker
+/*
+ErrMissingWorker is returned when a queue is created without a worker implementation.
+*/
 var ErrMissingWorker = errors.New("missing worker module")
 
-// NewQueue returns a Queue.
+// NewQueue creates and returns a new Queue instance with the provided options.
+// Returns an error if no worker is specified.
 func NewQueue(opts ...Option) (*Queue, error) {
 	o := NewOptions(opts...)
 	q := &Queue{
-		routineGroup:  newRoutineGroup(),
-		quit:          make(chan struct{}),
-		ready:         make(chan struct{}, 1),
-		notify:        make(chan struct{}, 1),
-		workerCount:   o.workerCount,
-		logger:        o.logger,
-		worker:        o.worker,
-		metric:        &metric{},
-		afterFn:       o.afterFn,
-		retryInterval: o.retryInterval,
+		routineGroup:  newRoutineGroup(),      // Manages all goroutines spawned by the queue
+		quit:          make(chan struct{}),    // Signals shutdown to all goroutines
+		ready:         make(chan struct{}, 1), // Signals when a worker is ready to process a job
+		notify:        make(chan struct{}, 1), // Notifies workers of new jobs
+		workerCount:   o.workerCount,          // Number of worker goroutines
+		logger:        o.logger,               // Logger for queue events
+		worker:        o.worker,               // Worker implementation
+		metric:        &metric{},              // Metrics collector
+		afterFn:       o.afterFn,              // Optional post-job callback
+		retryInterval: o.retryInterval,        // Interval for retrying job requests
 	}
 
 	if q.worker == nil {
@@ -61,7 +71,8 @@ func NewQueue(opts ...Option) (*Queue, error) {
 	return q, nil
 }
 
-// Start to enable all worker
+// Start launches all worker goroutines and begins processing jobs.
+// If workerCount is zero, Start is a no-op.
 func (q *Queue) Start() {
 	q.Lock()
 	count := q.workerCount
@@ -74,7 +85,9 @@ func (q *Queue) Start() {
 	})
 }
 
-// Shutdown stops all queues.
+// Shutdown initiates a graceful shutdown of the queue.
+// It signals all goroutines to stop, shuts down the worker, and closes the quit channel.
+// Shutdown is idempotent and safe to call multiple times.
 func (q *Queue) Shutdown() {
 	if !atomic.CompareAndSwapInt32(&q.stopFlag, 0, 1) {
 		return
@@ -92,55 +105,59 @@ func (q *Queue) Shutdown() {
 	})
 }
 
-// Release for graceful shutdown.
+// Release performs a graceful shutdown and waits for all goroutines to finish.
 func (q *Queue) Release() {
 	q.Shutdown()
 	q.Wait()
 }
 
-// BusyWorkers returns the numbers of workers in the running process.
+// BusyWorkers returns the number of workers currently processing jobs.
 func (q *Queue) BusyWorkers() int64 {
 	return q.metric.BusyWorkers()
 }
 
-// BusyWorkers returns the numbers of success tasks.
+// SuccessTasks returns the number of successfully completed tasks.
 func (q *Queue) SuccessTasks() uint64 {
 	return q.metric.SuccessTasks()
 }
 
-// BusyWorkers returns the numbers of failure tasks.
+// FailureTasks returns the number of failed tasks.
 func (q *Queue) FailureTasks() uint64 {
 	return q.metric.FailureTasks()
 }
 
-// BusyWorkers returns the numbers of submitted tasks.
+// SubmittedTasks returns the number of tasks submitted to the queue.
 func (q *Queue) SubmittedTasks() uint64 {
 	return q.metric.SubmittedTasks()
 }
 
-// CompletedTasks returns the numbers of completed tasks.
+// CompletedTasks returns the total number of completed tasks (success + failure).
 func (q *Queue) CompletedTasks() uint64 {
 	return q.metric.CompletedTasks()
 }
 
-// Wait all process
+// Wait blocks until all goroutines in the routine group have finished.
 func (q *Queue) Wait() {
 	q.routineGroup.Wait()
 }
 
-// Queue to queue single job with binary
+// Queue enqueues a single job (core.QueuedMessage) into the queue.
+// Accepts job options for customization.
 func (q *Queue) Queue(message core.QueuedMessage, opts ...job.AllowOption) error {
 	data := job.NewMessage(message, opts...)
 
 	return q.queue(&data)
 }
 
-// QueueTask to queue single task
+// QueueTask enqueues a single task function into the queue.
+// Accepts job options for customization.
 func (q *Queue) QueueTask(task job.TaskFunc, opts ...job.AllowOption) error {
 	data := job.NewTask(task, opts...)
 	return q.queue(&data)
 }
 
+// queue is an internal helper to enqueue a job.Message into the worker.
+// It increments the submitted task metric and notifies workers if possible.
 func (q *Queue) queue(m *job.Message) error {
 	if atomic.LoadInt32(&q.stopFlag) == 1 {
 		return ErrQueueShutdown
@@ -151,9 +168,8 @@ func (q *Queue) queue(m *job.Message) error {
 	}
 
 	q.metric.IncSubmittedTask()
-	// notify worker
-	// if the channel is full, it means that the worker is busy
-	// and we don't want to block the main thread
+	// Notify a worker that a new job is available.
+	// If the notify channel is full, the worker is busy and we avoid blocking.
 	select {
 	case q.notify <- struct{}{}:
 	default:
@@ -162,10 +178,11 @@ func (q *Queue) queue(m *job.Message) error {
 	return nil
 }
 
+// work executes a single task, handling panics and updating metrics accordingly.
+// After execution, it schedules the next worker if needed.
 func (q *Queue) work(task core.TaskMessage) {
 	var err error
-	// to handle panic cases from inside the worker
-	// in such case, we start a new goroutine
+	// Defer block to handle panics, update metrics, and run afterFn callback.
 	defer func() {
 		q.metric.DecBusyWorker()
 		e := recover()
@@ -174,7 +191,7 @@ func (q *Queue) work(task core.TaskMessage) {
 		}
 		q.schedule()
 
-		// increase success or failure number
+		// Update success or failure metrics based on execution result.
 		if err == nil && e == nil {
 			q.metric.IncSuccessTask()
 		} else {
@@ -190,6 +207,8 @@ func (q *Queue) work(task core.TaskMessage) {
 	}
 }
 
+// run dispatches the task to the appropriate handler based on its type.
+// Returns an error if the task type is invalid.
 func (q *Queue) run(task core.TaskMessage) error {
 	switch t := task.(type) {
 	case *job.Message:
@@ -199,8 +218,11 @@ func (q *Queue) run(task core.TaskMessage) error {
 	}
 }
 
+// handle executes a job.Message, supporting retries, timeouts, and panic recovery.
+// Returns an error if the job fails or times out.
 func (q *Queue) handle(m *job.Message) error {
-	// create channel with buffer size 1 to avoid goroutine leak
+	// done: receives the result of the job execution
+	// panicChan: receives any panic that occurs in the job goroutine
 	done := make(chan error, 1)
 	panicChan := make(chan any, 1)
 	startTime := time.Now()
@@ -209,18 +231,18 @@ func (q *Queue) handle(m *job.Message) error {
 		cancel()
 	}()
 
-	// run the job
+	// Run the job in a separate goroutine to support timeout and panic recovery.
 	go func() {
-		// handle panic issue
+		// Defer block to catch panics and send to panicChan
 		defer func() {
 			if p := recover(); p != nil {
 				panicChan <- p
 			}
 		}()
 
-		// run custom process function
 		var err error
 
+		// Set up backoff for retry logic
 		b := &backoff.Backoff{
 			Min:    m.RetryMin,
 			Max:    m.RetryMax,
@@ -230,26 +252,28 @@ func (q *Queue) handle(m *job.Message) error {
 		delay := m.RetryDelay
 	loop:
 		for {
+			// If a custom Task function is provided, use it; otherwise, use the worker's Run method.
 			if m.Task != nil {
 				err = m.Task(ctx)
 			} else {
 				err = q.worker.Run(ctx, m)
 			}
 
-			// check error and retry count
+			// If no error or no retries left, exit loop.
 			if err == nil || m.RetryCount == 0 {
 				break
 			}
 			m.RetryCount--
 
+			// If no fixed retry delay, use backoff.
 			if m.RetryDelay == 0 {
 				delay = b.Duration()
 			}
 
 			select {
-			case <-time.After(delay): // retry delay
+			case <-time.After(delay): // Wait before retrying
 				q.logger.Infof("retry remaining times: %d, delay time: %s", m.RetryCount, delay)
-			case <-ctx.Done(): // timeout reached
+			case <-ctx.Done(): // Timeout reached
 				err = ctx.Err()
 				break loop
 			}
@@ -261,36 +285,36 @@ func (q *Queue) handle(m *job.Message) error {
 	select {
 	case p := <-panicChan:
 		panic(p)
-	case <-ctx.Done(): // timeout reached
+	case <-ctx.Done(): // Timeout reached
 		return ctx.Err()
-	case <-q.quit: // shutdown service
-		// cancel job
+	case <-q.quit: // Queue is shutting down
+		// Cancel job and wait for remaining time or job completion
 		cancel()
-
 		leftTime := m.Timeout - time.Since(startTime)
-		// wait job
 		select {
 		case <-time.After(leftTime):
 			return context.DeadlineExceeded
-		case err := <-done: // job finish
+		case err := <-done: // Job finished
 			return err
 		case p := <-panicChan:
 			panic(p)
 		}
-	case err := <-done: // job finish
+	case err := <-done: // Job finished
 		return err
 	}
 }
 
-// UpdateWorkerCount to update worker number dynamically.
+// UpdateWorkerCount dynamically updates the number of worker goroutines.
+// Triggers scheduling to adjust to the new worker count.
 func (q *Queue) UpdateWorkerCount(num int64) {
 	q.Lock()
 	q.workerCount = num
 	q.Unlock()
 	q.schedule()
 }
 
-// schedule to check worker number
+// schedule checks if more workers can be started based on the current busy count.
+// If so, it signals readiness to start a new worker.
 func (q *Queue) schedule() {
 	q.Lock()
 	defer q.Unlock()
@@ -304,24 +328,30 @@ func (q *Queue) schedule() {
 	}
 }
 
-// start to start all worker
+/*
+start launches the main worker loop, which manages job scheduling and execution.
+
+- It uses a ticker to periodically retry job requests if the queue is empty.
+- For each available worker slot, it requests a new task from the worker.
+- If a task is available, it is sent to the tasks channel and processed by a new goroutine.
+- The loop exits when the quit channel is closed.
+*/
 func (q *Queue) start() {
 	tasks := make(chan core.TaskMessage, 1)
 	ticker := time.NewTicker(q.retryInterval)
 	defer ticker.Stop()
 
 	for {
-		// check worker number
+		// Ensure the number of busy workers does not exceed the configured worker count.
 		q.schedule()
 
 		select {
-		// wait worker ready
-		case <-q.ready:
-		case <-q.quit:
+		case <-q.ready: // Wait for a worker slot to become available
+		case <-q.quit: // Shutdown signal received
 			return
 		}
 
-		// request task from queue in background
+		// Request a task from the worker in a background goroutine.
 		q.routineGroup.Run(func() {
 			for {
 				t, err := q.worker.Request()
@@ -359,7 +389,7 @@ func (q *Queue) start() {
 			return
 		}
 
-		// start new task
+		// Start processing the new task in a separate goroutine.
 		q.metric.IncBusyWorker()
 		q.routineGroup.Run(func() {
 			q.work(task)