feat(queue): implement global rate limiting and robust state management

- Replace per-worker rate limiters with a single global rate limiter
- Add atomic file operations for state persistence
- Implement backup and recovery mechanisms for queue state
- Add temporary file handling during state saves
- Improve error handling and logging for state operations

BREAKING: none

Author: teslashibe

pkg/x/queue.go

@@ -89,7 +89,6 @@ type Worker struct {
 	id         int              // Unique identifier for the worker
 	jobChannel chan RequestData // Channel for receiving jobs
 	quit       chan bool        // Channel for shutdown signaling
-	rateLimit  *time.Ticker     // Rate limiter for API requests
 	queue      *RequestQueue    // Reference to parent queue
 }
 
@@ -103,6 +102,7 @@ type RequestQueue struct {
 	mu                sync.Mutex                     // Mutex for thread-safe operations
 	paused            bool                           // Pause state for queue processing
 	pauseMux          sync.RWMutex                   // Mutex for pause state access
+	globalRateLimit   *time.Ticker                   // Global rate limiter shared across workers
 }
 
 // NewRequestQueue creates and initializes a new RequestQueue instance
@@ -122,6 +122,7 @@ func NewRequestQueue(maxWorkers int) *RequestQueue {
 		maxWorkers:        maxWorkers,
 		jobChannel:        make(chan RequestData, maxWorkers*2),
 		requestsPerSecond: DefaultAPIRequestsPerSecond,
+		globalRateLimit:   time.NewTicker(time.Second / DefaultAPIRequestsPerSecond),
 	}
 
 	// Initialize queues
@@ -231,7 +232,6 @@ func (rq *RequestQueue) newWorker(id int) *Worker {
 		id:         id,
 		jobChannel: rq.jobChannel,
 		quit:       make(chan bool),
-		rateLimit:  time.NewTicker(time.Second / time.Duration(rq.requestsPerSecond)),
 		queue:      rq,
 	}
 }
@@ -331,7 +331,7 @@ func (w *Worker) start() {
 	for {
 		select {
 		case job := <-w.jobChannel:
-			<-w.rateLimit.C // Rate limiting
+			<-w.queue.globalRateLimit.C // Global rate limiting
 			w.processRequest(job)
 		case <-w.quit:
 			return
@@ -459,7 +459,7 @@ func (rq *RequestQueue) Stop() {
 
 	// Wait for workers to finish
 	for _, worker := range rq.workers {
-		worker.rateLimit.Stop()
+		worker.queue.globalRateLimit.Stop()
 	}
 
 	// Close job channel after all workers have stopped
@@ -487,7 +487,7 @@ func (rq *RequestQueue) StopWithContext(ctx context.Context) error {
 		// Force close channels
 		close(rq.jobChannel)
 		for _, worker := range rq.workers {
-			worker.rateLimit.Stop()
+			worker.queue.globalRateLimit.Stop()
 			close(worker.quit)
 		}
 		return ctx.Err()
@@ -557,10 +557,6 @@ func (rq *RequestQueue) SetRequestsPerSecond(rps float64) {
 	rq.requestsPerSecond = rps
 	logger.Infof("Queue rate limit set to %.2f requests per second", rps)
 
-	// Update existing workers' rate limiters
-	if rq.workers != nil {
-		for _, worker := range rq.workers {
-			worker.rateLimit.Reset(time.Second / time.Duration(rps))
-		}
-	}
+	// Update the global rate limiter
+	rq.globalRateLimit.Reset(time.Second / time.Duration(rps))
 }

pkg/x/queue_state.go

@@ -76,14 +76,30 @@ func (rq *RequestQueue) SaveState() error {
 		return err
 	}
 
-	// Write state to file
+	// Write state to temporary file first
+	statePath := filepath.Join(stateDir, queueStateFile)
+	tempPath := statePath + ".tmp"
+
 	data, err := json.MarshalIndent(state, "", "  ")
 	if err != nil {
 		return err
 	}
 
-	statePath := filepath.Join(stateDir, queueStateFile)
-	if err := os.WriteFile(statePath, data, 0644); err != nil {
+	// Write to temp file first
+	if err := os.WriteFile(tempPath, data, 0644); err != nil {
+		return err
+	}
+
+	// Backup existing state file if it exists
+	if _, err := os.Stat(statePath); err == nil {
+		backupPath := statePath + ".bak"
+		if err := os.Rename(statePath, backupPath); err != nil {
+			logger.Warnf("Failed to create backup file: %v", err)
+		}
+	}
+
+	// Atomically rename temp file to actual state file
+	if err := os.Rename(tempPath, statePath); err != nil {
 		return err
 	}
 
@@ -94,13 +110,26 @@ func (rq *RequestQueue) SaveState() error {
 // LoadState loads the queue state from disk
 func (rq *RequestQueue) LoadState() error {
 	statePath := filepath.Join(getStateDir(), queueStateFile)
+
+	// Try loading the main state file
 	data, err := os.ReadFile(statePath)
 	if err != nil {
-		if os.IsNotExist(err) {
+		if !os.IsNotExist(err) {
+			// Try loading from backup if main file is corrupted
+			backupPath := statePath + ".bak"
+			data, err = os.ReadFile(backupPath)
+			if err != nil {
+				if os.IsNotExist(err) {
+					logger.Debugf("No existing queue state found at %s", statePath)
+					return nil
+				}
+				return err
+			}
+			logger.Warnf("Loaded queue state from backup file")
+		} else {
 			logger.Debugf("No existing queue state found at %s", statePath)
 			return nil
 		}
-		return err
 	}
 
 	var state QueueState