background runner

2025-08-12 06:37:16 -07:00
parent a8e1095abc
commit 7a693d42f5
13 changed files with 2817 additions and 0 deletions
@@ -0,0 +1,254 @@
+package backgroundtask
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// Runner executes a work function periodically or when signaled.
+// It provides graceful shutdown, work coalescing, and statistics tracking.
+type Runner struct {
+	config       map[string]any // Configuration passed to work function
+	logger       *slog.Logger   // Logger for output
+	workFunc     WorkFunc       // The function to execute
+	wakeCh       chan struct{}  // Channel to signal immediate work
+	stopCh       chan struct{}  // Channel to signal shutdown
+	doneCh       chan struct{}  // Channel closed when goroutine exits
+	ticker       *time.Ticker   // Periodic timer
+	mu           sync.Mutex     // Protects started flag
+	started      bool           // Whether Run has been called
+	stopped      atomic.Bool    // Whether shutdown has been called
+	working      atomic.Bool    // Whether work is currently executing
+	pendingWork  atomic.Bool    // Whether work was requested during execution
+	interval     time.Duration  // Time between periodic executions
+	stats        Stats          // Execution statistics
+	shutdownOnce sync.Once      // Ensures shutdown runs once
+}
+
+// Initialize creates a new Runner with the provided configuration, logger, and work function.
+// Options can be provided to customize the runner's behavior.
+func Initialize(config map[string]any, logger *slog.Logger, workFunc WorkFunc, opts ...Option) (*Runner, error) {
+	if config == nil {
+		return nil, ErrNilConfig
+	}
+	if workFunc == nil {
+		return nil, ErrNilWorkFunc
+	}
+	if logger == nil {
+		logger = slog.Default()
+	}
+
+	r := &Runner{
+		config:   config,
+		logger:   logger,
+		workFunc: workFunc,
+		wakeCh:   make(chan struct{}, 1),
+		stopCh:   make(chan struct{}),
+		doneCh:   make(chan struct{}),
+		interval: DefaultInterval,
+	}
+
+	// Apply options
+	for _, opt := range opts {
+		if err := opt(r); err != nil {
+			return nil, err
+		}
+	}
+
+	return r, nil
+}
+
+// Stats returns a pointer to the current statistics.
+// This method is thread-safe and can be called concurrently.
+func (r *Runner) Stats() *Stats {
+	return &r.stats
+}
+
+// Run starts the background goroutine that executes the work function.
+// This method is idempotent - calling it multiple times has no effect if already running.
+// Returns ErrAlreadyRunning if called after shutdown.
+func (r *Runner) Run() error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	if r.stopped.Load() {
+		return ErrStopped
+	}
+
+	if r.started {
+		// Idempotent - already running
+		return nil
+	}
+
+	r.started = true
+	r.ticker = time.NewTicker(r.interval)
+
+	// Start the worker goroutine
+	go r.worker()
+
+	r.logger.Info("background runner started",
+		"interval", r.interval,
+		"config_type", r.config != nil,
+	)
+
+	return nil
+}
+
+// Signal requests immediate execution of the work function.
+// If the runner is already executing work, the signal is coalesced and
+// another execution will occur after the current one completes.
+// Returns ErrStopped if the runner has been shut down.
+func (r *Runner) Signal() error {
+	if r.stopped.Load() {
+		return ErrStopped
+	}
+
+	// Non-blocking send - if channel is full, work is already pending
+	select {
+	case r.wakeCh <- struct{}{}:
+		r.logger.Debug("work signaled")
+	default:
+		// Channel full - work already pending
+		// Set pendingWork flag to ensure we run again after current work
+		r.pendingWork.Store(true)
+		r.logger.Debug("work signal coalesced")
+	}
+
+	return nil
+}
+
+// Shutdown gracefully stops the runner, waiting for any in-progress work to complete.
+// The provided context controls the shutdown timeout. If the context has no deadline,
+// a default timeout of MaxShutdownTimeout (60 seconds) is applied.
+// This method is idempotent - multiple calls will not cause issues.
+func (r *Runner) Shutdown(ctx context.Context) error {
+	var err error
+
+	r.shutdownOnce.Do(func() {
+		r.logger.Info("shutting down background runner")
+
+		// Mark as stopped to prevent new signals
+		r.stopped.Store(true)
+
+		// Stop the ticker
+		if r.ticker != nil {
+			r.ticker.Stop()
+		}
+
+		// Signal the worker to stop
+		close(r.stopCh)
+
+		// Apply default timeout if context has no deadline
+		if _, hasDeadline := ctx.Deadline(); !hasDeadline {
+			var cancel context.CancelFunc
+			ctx, cancel = context.WithTimeout(ctx, MaxShutdownTimeout)
+			defer cancel()
+		}
+
+		// Wait for worker to finish or timeout
+		select {
+		case <-r.doneCh:
+			r.logger.Info("background runner shutdown complete")
+		case <-ctx.Done():
+			r.logger.Error("background runner shutdown timeout")
+			err = ErrShutdownTimeout
+		}
+	})
+
+	return err
+}
+
+// worker is the main goroutine that executes the work function.
+func (r *Runner) worker() {
+	defer close(r.doneCh)
+	defer r.ticker.Stop()
+
+	r.logger.Debug("worker started")
+
+	for {
+		select {
+		case <-r.wakeCh:
+			r.logger.Debug("worker woken by signal")
+			r.tryWork("signal")
+
+		case <-r.ticker.C:
+			r.logger.Debug("worker woken by ticker")
+			r.tryWork("ticker")
+
+		case <-r.stopCh:
+			r.logger.Debug("worker stopping")
+			return
+		}
+	}
+}
+
+// tryWork attempts to execute the work function, handling coalescing and pending work.
+func (r *Runner) tryWork(trigger string) {
+	// Check if already working
+	if !r.working.CompareAndSwap(false, true) {
+		// Already working - set pending flag
+		r.pendingWork.Store(true)
+		r.logger.Debug("work already in progress, pending flag set", "trigger", trigger)
+		return
+	}
+
+	// Execute work
+	r.doWork(trigger)
+
+	// Clear working flag
+	r.working.Store(false)
+
+	// Check if there's pending work
+	if r.pendingWork.CompareAndSwap(true, false) {
+		r.logger.Debug("processing pending work")
+		// Recursively process pending work (will only recurse once due to working flag)
+		r.tryWork("pending")
+	}
+}
+
+// doWork executes the actual work function with proper context and stats tracking.
+func (r *Runner) doWork(trigger string) {
+	// Create context that will be canceled on shutdown
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// Create a done channel for this work execution
+	done := make(chan struct{})
+	defer close(done)
+
+	// Monitor for shutdown
+	go func() {
+		select {
+		case <-r.stopCh:
+			cancel()
+		case <-done:
+			// Work completed
+		}
+	}()
+
+	// Update stats
+	r.stats.enterWork()
+	defer r.stats.exitWork()
+
+	startTime := time.Now()
+	r.stats.setLastStart(startTime)
+
+	r.logger.Debug("executing work", "trigger", trigger)
+
+	// Execute the work function
+	err := r.workFunc(ctx, r.logger, r.config)
+
+	// Update stats
+	r.stats.setLastFinish(time.Now())
+	r.stats.incrementWorkCycles()
+	if err != nil {
+		r.stats.setLastErr(err)
+		r.logger.Error("work function error", "error", err, "duration", time.Since(startTime))
+	} else {
+		r.stats.setLastErr(nil)
+		r.logger.Debug("work completed", "duration", time.Since(startTime))
+	}
+}