refactor: 优化优雅关机和完善失败请求处理

- 使用 channel 通信替代定时轮询，请求完成时实时打印日志
- 标记失败请求时设置 end_time 和计算 duration_ms
- 同步处理 upstream_attempt 表的失败状态

Author: Bowl42

cmd/maxx/main.go

@@ -121,6 +121,12 @@ func main() {
 	} else if count > 0 {
 		log.Printf("Marked %d stale requests as failed", count)
 	}
+	// Also mark stale upstream attempts as failed
+	if count, err := attemptRepo.MarkStaleAttemptsFailed(); err != nil {
+		log.Printf("Warning: Failed to mark stale attempts: %v", err)
+	} else if count > 0 {
+		log.Printf("Marked %d stale upstream attempts as failed", count)
+	}
 
 	// Create cached repositories
 	cachedProviderRepo := cached.NewProviderRepository(providerRepo)
@@ -344,15 +350,15 @@ func main() {
 	activeCount := requestTracker.ActiveCount()
 	if activeCount > 0 {
 		log.Printf("Waiting for %d active proxy requests to complete...", activeCount)
-		completed := requestTracker.GracefulShutdown(core.GracefulShutdownTimeout, core.GracefulShutdownCheckInterval)
+		completed := requestTracker.GracefulShutdown(core.GracefulShutdownTimeout)
 		if !completed {
 			log.Printf("Graceful shutdown timeout, some requests may be interrupted")
 		} else {
 			log.Printf("All proxy requests completed successfully")
 		}
 	} else {
 		// Mark as shutting down to reject new requests
-		requestTracker.GracefulShutdown(0, time.Second)
+		requestTracker.GracefulShutdown(0)
 		log.Printf("No active proxy requests")
 	}
 

internal/core/database.go

@@ -172,6 +172,12 @@ func InitializeServerComponents(
 	} else if count > 0 {
 		log.Printf("[Core] Marked %d stale requests as failed", count)
 	}
+	// Also mark stale upstream attempts as failed
+	if count, err := repos.AttemptRepo.MarkStaleAttemptsFailed(); err != nil {
+		log.Printf("[Core] Warning: Failed to mark stale attempts: %v", err)
+	} else if count > 0 {
+		log.Printf("[Core] Marked %d stale upstream attempts as failed", count)
+	}
 
 	log.Printf("[Core] Loading cached data")
 	if err := repos.CachedProviderRepo.Load(); err != nil {

internal/core/request_tracker.go

@@ -14,6 +14,9 @@ type RequestTracker struct {
 	wg          sync.WaitGroup
 	shutdownCh  chan struct{}
 	isShutdown  atomic.Bool
+	// notifyCh is used to notify when a request completes during shutdown
+	notifyCh chan struct{}
+	notifyMu sync.Mutex
 }
 
 // NewRequestTracker creates a new request tracker
@@ -36,8 +39,23 @@ func (t *RequestTracker) Add() bool {
 
 // Done decrements the active request count
 func (t *RequestTracker) Done() {
-	atomic.AddInt64(&t.activeCount, -1)
+	remaining := atomic.AddInt64(&t.activeCount, -1)
 	t.wg.Done()
+
+	// Notify shutdown goroutine if shutting down
+	if t.isShutdown.Load() {
+		t.notifyMu.Lock()
+		ch := t.notifyCh
+		t.notifyMu.Unlock()
+		if ch != nil {
+			select {
+			case ch <- struct{}{}:
+			default:
+				// Non-blocking send, channel might be full or closed
+			}
+		}
+		log.Printf("[RequestTracker] Request completed, %d remaining", remaining)
+	}
 }
 
 // ActiveCount returns the current number of active requests
@@ -97,8 +115,12 @@ func (t *RequestTracker) ShutdownCh() <-chan struct{} {
 
 // GracefulShutdown initiates graceful shutdown and waits for requests to complete
 // maxWait: maximum time to wait for requests to complete
-// checkInterval: how often to log remaining requests
-func (t *RequestTracker) GracefulShutdown(maxWait time.Duration, checkInterval time.Duration) bool {
+func (t *RequestTracker) GracefulShutdown(maxWait time.Duration) bool {
+	// Setup notify channel before marking shutdown
+	t.notifyMu.Lock()
+	t.notifyCh = make(chan struct{}, 100) // Buffered to avoid blocking Done()
+	t.notifyMu.Unlock()
+
 	t.isShutdown.Store(true)
 	close(t.shutdownCh)
 
@@ -116,19 +138,21 @@ func (t *RequestTracker) GracefulShutdown(maxWait time.Duration, checkInterval t
 		close(done)
 	}()
 
-	ticker := time.NewTicker(checkInterval)
-	defer ticker.Stop()
-
 	deadline := time.After(maxWait)
 
 	for {
 		select {
 		case <-done:
 			log.Printf("[RequestTracker] All requests completed, shutdown clean")
 			return true
-		case <-ticker.C:
-			remaining := t.ActiveCount()
-			log.Printf("[RequestTracker] Waiting for %d active requests to complete...", remaining)
+		case <-t.notifyCh:
+			// Request completed notification received, log is printed in Done()
+			// Check if all done
+			if t.ActiveCount() == 0 {
+				<-done // Wait for wg.Wait() to complete
+				log.Printf("[RequestTracker] All requests completed, shutdown clean")
+				return true
+			}
 		case <-deadline:
 			remaining := t.ActiveCount()
 			log.Printf("[RequestTracker] Timeout reached, %d requests still active, forcing shutdown", remaining)

internal/core/server.go

@@ -13,8 +13,6 @@ import (
 const (
 	// GracefulShutdownTimeout is the maximum time to wait for active requests
 	GracefulShutdownTimeout = 2 * time.Minute
-	// GracefulShutdownCheckInterval is how often to log remaining requests
-	GracefulShutdownCheckInterval = 5 * time.Second
 	// HTTPShutdownTimeout is the timeout for HTTP server shutdown after requests complete
 	HTTPShutdownTimeout = 5 * time.Second
 )
@@ -136,16 +134,15 @@ func (s *ManagedServer) Stop(ctx context.Context) error {
 		if activeCount > 0 {
 			log.Printf("[Server] Waiting for %d active proxy requests to complete...", activeCount)
 
-			// Use GracefulShutdown with logging
-			completed := tracker.GracefulShutdown(GracefulShutdownTimeout, GracefulShutdownCheckInterval)
+			completed := tracker.GracefulShutdown(GracefulShutdownTimeout)
 			if !completed {
 				log.Printf("[Server] Graceful shutdown timeout, some requests may be interrupted")
 			} else {
 				log.Printf("[Server] All proxy requests completed successfully")
 			}
 		} else {
 			// Mark as shutting down to reject new requests
-			tracker.GracefulShutdown(0, time.Second)
+			tracker.GracefulShutdown(0)
 			log.Printf("[Server] No active proxy requests")
 		}
 	}

internal/repository/interfaces.go

@@ -116,6 +116,8 @@ type ProxyUpstreamAttemptRepository interface {
 	UpdateCost(id uint64, cost uint64) error
 	// BatchUpdateCosts updates costs for multiple attempts in a single transaction
 	BatchUpdateCosts(updates map[uint64]uint64) error
+	// MarkStaleAttemptsFailed marks stale attempts as failed with proper end_time and duration
+	MarkStaleAttemptsFailed() (int64, error)
 }
 
 type SystemSettingRepository interface {

internal/repository/sqlite/proxy_request.go

@@ -150,25 +150,32 @@ func (r *ProxyRequestRepository) CountWithFilter(filter *repository.ProxyRequest
 
 // MarkStaleAsFailed marks all IN_PROGRESS/PENDING requests from other instances as FAILED
 // Also marks requests that have been IN_PROGRESS for too long (> 30 minutes) as timed out
+// Sets proper end_time and duration_ms for complete failure handling
 func (r *ProxyRequestRepository) MarkStaleAsFailed(currentInstanceID string) (int64, error) {
 	timeoutThreshold := time.Now().Add(-30 * time.Minute).UnixMilli()
 	now := time.Now().UnixMilli()
 
 	// Use raw SQL for complex CASE expression
+	// Sets end_time = now and calculates duration_ms = now - start_time
 	result := r.db.gorm.Exec(`
 		UPDATE proxy_requests
 		SET status = 'FAILED',
 		    error = CASE
 		        WHEN instance_id IS NULL OR instance_id != ? THEN 'Server restarted'
 		        ELSE 'Request timed out (stuck in progress)'
 		    END,
+		    end_time = ?,
+		    duration_ms = CASE
+		        WHEN start_time > 0 THEN ? - start_time
+		        ELSE 0
+		    END,
 		    updated_at = ?
 		WHERE status IN ('PENDING', 'IN_PROGRESS')
 		  AND (
 		      (instance_id IS NULL OR instance_id != ?)
 		      OR (start_time < ? AND start_time > 0)
 		  )`,
-		currentInstanceID, now, currentInstanceID, timeoutThreshold,
+		currentInstanceID, now, now, now, currentInstanceID, timeoutThreshold,
 	)
 	if result.Error != nil {
 		return 0, result.Error

internal/repository/sqlite/proxy_upstream_attempt.go

@@ -133,6 +133,34 @@ func (r *ProxyUpstreamAttemptRepository) UpdateCost(id uint64, cost uint64) erro
 	return r.db.gorm.Model(&ProxyUpstreamAttempt{}).Where("id = ?", id).Update("cost", cost).Error
 }
 
+// MarkStaleAttemptsFailed marks all IN_PROGRESS/PENDING attempts belonging to stale requests as FAILED
+// This should be called after MarkStaleAsFailed on proxy_requests to clean up orphaned attempts
+// Sets proper end_time and duration_ms for complete failure handling
+func (r *ProxyUpstreamAttemptRepository) MarkStaleAttemptsFailed() (int64, error) {
+	now := time.Now().UnixMilli()
+
+	// Update attempts that belong to FAILED requests but are still in progress
+	result := r.db.gorm.Exec(`
+		UPDATE proxy_upstream_attempts
+		SET status = 'FAILED',
+		    end_time = ?,
+		    duration_ms = CASE
+		        WHEN start_time > 0 THEN ? - start_time
+		        ELSE 0
+		    END,
+		    updated_at = ?
+		WHERE status IN ('PENDING', 'IN_PROGRESS')
+		  AND proxy_request_id IN (
+		      SELECT id FROM proxy_requests WHERE status = 'FAILED'
+		  )`,
+		now, now, now,
+	)
+	if result.Error != nil {
+		return 0, result.Error
+	}
+	return result.RowsAffected, nil
+}
+
 // BatchUpdateCosts updates costs for multiple attempts in a single transaction
 func (r *ProxyUpstreamAttemptRepository) BatchUpdateCosts(updates map[uint64]uint64) error {
 	if len(updates) == 0 {