Week 1 Day 6：Retry / Backoff / Idempotency - 苏格拉底教学#

backoff := baseDelay * factor^attempt
backoff = backoff * (0.5 + rand.Float64()*0.5) // 加50%-100%随机

select {
case <-time.After(backoff):
    // 继续重试
case <-ctx.Done():
    return ctx.Err() // 立即退出
}

// internal/retry/errors.go
package retry

import (
	"context"
	"errors"
	"net"
	"net/http"
	"strings"
)

// IsRetryable 判断错误是否可重试
func IsRetryable(err error) bool {
	if err == nil {
		return false
	}

	// Context取消/超时：区分处理
	if errors.Is(err, context.Canceled) {
		return false // 用户主动取消
	}
	if errors.Is(err, context.DeadlineExceeded) {
		// 注意：这里判断的是"传入的ctx超时"还是"单次rpc超时"
		// 如果是外层ctx超时，不该重试；如果是内层单次超时，可重试
		// 实战中用专用error包装
		return false
	}

	// 网络错误
	var netErr net.Error
	if errors.As(err, &netErr) {
		if netErr.Timeout() {
			return true
		}
	}

	// OpenAI SDK的HTTPStatusCode
	var apiErr *HTTPError
	if errors.As(err, &apiErr) {
		return isRetryableStatus(apiErr.StatusCode)
	}

	// 字符串fallback（不推荐，但真实世界很常见）
	msg := err.Error()
	if strings.Contains(msg, "connection reset") ||
		strings.Contains(msg, "broken pipe") {
		return true
	}

	return false
}

type HTTPError struct {
	StatusCode int
	Body       string
}

func (e *HTTPError) Error() string {
	return e.Body
}

func isRetryableStatus(code int) bool {
	switch code {
	case http.StatusTooManyRequests: // 429
		return true
	case http.StatusInternalServerError, // 500
		http.StatusBadGateway,          // 502
		http.StatusServiceUnavailable,  // 503
		http.StatusGatewayTimeout:      // 504
		return true
	}
	return false
}

// internal/retry/backoff.go
package retry

import (
	"context"
	"fmt"
	"math/rand"
	"time"
)

type Config struct {
	MaxAttempts int           // 最大尝试次数（含首次）
	BaseDelay   time.Duration // 初始延迟
	MaxDelay    time.Duration // 单次最大延迟
	Factor      float64       // 递增因子（推荐1.5）
	Jitter      bool          // 是否加随机
}

func Default() Config {
	return Config{
		MaxAttempts: 3,
		BaseDelay:   500 * time.Millisecond,
		MaxDelay:    30 * time.Second,
		Factor:      1.5,
		Jitter:      true,
	}
}

// Do 执行op，按config重试
func Do(ctx context.Context, cfg Config, op func(ctx context.Context) error) error {
	var lastErr error
	delay := cfg.BaseDelay

	for attempt := 0; attempt < cfg.MaxAttempts; attempt++ {
		// 检查ctx是否已取消
		if err := ctx.Err(); err != nil {
			return err
		}

		err := op(ctx)
		if err == nil {
			return nil
		}
		lastErr = err

		// 不可重试，立即返回
		if !IsRetryable(err) {
			return fmt.Errorf("non-retryable: %w", err)
		}

		// 最后一次尝试失败，不再sleep
		if attempt == cfg.MaxAttempts-1 {
			break
		}

		// 计算等待时间
		sleep := delay
		if cfg.Jitter {
			// jitter: 0.5x - 1.5x of delay
			sleep = time.Duration(float64(delay) * (0.5 + rand.Float64()))
		}

		// 等待，同时监听ctx
		select {
		case <-time.After(sleep):
		case <-ctx.Done():
			return ctx.Err()
		}

		// 下次延迟
		delay = time.Duration(float64(delay) * cfg.Factor)
		if delay > cfg.MaxDelay {
			delay = cfg.MaxDelay
		}
	}

	return fmt.Errorf("exhausted %d attempts, last error: %w", cfg.MaxAttempts, lastErr)
}

err := retry.Do(ctx, retry.Default(), func(ctx context.Context) error {
	return llmClient.Call(ctx, req)
})

// internal/idempotency/store.go
package idempotency

import (
	"context"
	"encoding/json"
	"errors"
	"sync"
	"time"
)

var ErrInProgress = errors.New("request in progress")

type Entry struct {
	Result    json.RawMessage
	Err       string // 空字符串表示成功
	CreatedAt time.Time
	Status    string // "pending" | "done"
}

type MemStore struct {
	mu   sync.Mutex
	data map[string]*Entry
	ttl  time.Duration
}

func NewMemStore(ttl time.Duration) *MemStore {
	s := &MemStore{
		data: make(map[string]*Entry),
		ttl:  ttl,
	}
	go s.gc()
	return s
}

// AcquireOrGet 占坑或返回已有结果
// 返回 (entry, isNew)。isNew=true表示本goroutine应该执行实际操作
func (s *MemStore) AcquireOrGet(key string) (*Entry, bool) {
	s.mu.Lock()
	defer s.mu.Unlock()

	if e, ok := s.data[key]; ok {
		return e, false
	}
	e := &Entry{Status: "pending", CreatedAt: time.Now()}
	s.data[key] = e
	return e, true
}

// Complete 写入最终结果
func (s *MemStore) Complete(key string, result json.RawMessage, opErr error) {
	s.mu.Lock()
	defer s.mu.Unlock()

	e, ok := s.data[key]
	if !ok {
		return
	}
	e.Result = result
	e.Status = "done"
	if opErr != nil {
		e.Err = opErr.Error()
	}
}

func (s *MemStore) gc() {
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()
	for range ticker.C {
		s.mu.Lock()
		for k, e := range s.data {
			if time.Since(e.CreatedAt) > s.ttl {
				delete(s.data, k)
			}
		}
		s.mu.Unlock()
	}
}

func CreateTicket(ctx context.Context, store *idempotency.MemStore, key string, req TicketReq) (*TicketResp, error) {
	entry, isNew := store.AcquireOrGet(key)

	if !isNew {
		// 已有结果
		if entry.Status == "done" {
			if entry.Err != "" {
				return nil, errors.New(entry.Err)
			}
			var resp TicketResp
			json.Unmarshal(entry.Result, &resp)
			return &resp, nil
		}
		// pending，说明另一goroutine在处理——取决于策略：等、或者拒绝
		return nil, idempotency.ErrInProgress
	}

	// 新请求，实际创建
	resp, err := actuallyCreateTicket(ctx, req)
	raw, _ := json.Marshal(resp)
	store.Complete(key, raw, err)
	return resp, err
}

// internal/idempotency/redis_store.go
package idempotency

import (
	"context"
	"encoding/json"
	"time"

	"github.com/redis/go-redis/v9"
)

type RedisStore struct {
	rdb *redis.Client
	ttl time.Duration
}

func NewRedisStore(rdb *redis.Client, ttl time.Duration) *RedisStore {
	return &RedisStore{rdb: rdb, ttl: ttl}
}

// TryAcquire 原子占坑；返回true表示获得锁
func (s *RedisStore) TryAcquire(ctx context.Context, key string) (bool, error) {
	// SETNX: set if not exists
	ok, err := s.rdb.SetNX(ctx, "idem:"+key, "pending", s.ttl).Result()
	return ok, err
}

// Get 查询已存结果
func (s *RedisStore) Get(ctx context.Context, key string) (*Entry, error) {
	val, err := s.rdb.Get(ctx, "idem:"+key).Bytes()
	if err == redis.Nil {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	if string(val) == "pending" {
		return &Entry{Status: "pending"}, nil
	}
	var e Entry
	if err := json.Unmarshal(val, &e); err != nil {
		return nil, err
	}
	return &e, nil
}

// Complete 覆盖为最终结果
func (s *RedisStore) Complete(ctx context.Context, key string, result json.RawMessage, opErr error) error {
	e := Entry{Result: result, Status: "done", CreatedAt: time.Now()}
	if opErr != nil {
		e.Err = opErr.Error()
	}
	raw, _ := json.Marshal(e)
	return s.rdb.Set(ctx, "idem:"+key, raw, s.ttl).Err()
}

func CreateTicketIdempotent(ctx context.Context, s *idempotency.RedisStore, key string, req TicketReq) (*TicketResp, error) {
	// 1. 查是否已有结果
	if e, err := s.Get(ctx, key); err == nil && e != nil && e.Status == "done" {
		if e.Err != "" {
			return nil, errors.New(e.Err)
		}
		var resp TicketResp
		json.Unmarshal(e.Result, &resp)
		return &resp, nil
	}

	// 2. 原子占坑
	acquired, err := s.TryAcquire(ctx, key)
	if err != nil {
		return nil, err
	}
	if !acquired {
		// 别的goroutine在处理或刚完成；轮询等待
		return waitForResult(ctx, s, key, 10*time.Second)
	}

	// 3. 执行并写回
	resp, opErr := actuallyCreateTicket(ctx, req)
	raw, _ := json.Marshal(resp)
	s.Complete(ctx, key, raw, opErr)

	return resp, opErr
}

// internal/idempotency/key.go
package idempotency

import (
	"crypto/sha256"
	"encoding/hex"
	"fmt"
)

// KeyFor 为 create_ticket 生成key
// 业务语义：同一user的同一request_id在一段窗口内视为同一操作
func KeyFor(userID, requestID, operation string) string {
	raw := fmt.Sprintf("%s|%s|%s", userID, requestID, operation)
	h := sha256.Sum256([]byte(raw))
	return hex.EncodeToString(h[:])
}

func (a *Agent) CreateTicketSafe(ctx context.Context, userID, reqID string, req TicketReq) (*TicketResp, error) {
	key := idempotency.KeyFor(userID, reqID, "create_ticket")

	var resp *TicketResp
	err := retry.Do(ctx, retry.Default(), func(ctx context.Context) error {
		r, err := CreateTicketIdempotent(ctx, a.store, key, req)
		if err != nil {
			return err
		}
		resp = r
		return nil
	})
	return resp, err
}

ctx, cancel := context.WithTimeout(parentCtx, 3*time.Second)
defer cancel()

var ErrRPCTimeout = errors.New("single rpc timeout")
if errors.Is(err, ErrRPCTimeout) { return true }

// 判断课程依赖图是否有环
func canFinish(numCourses int, prerequisites [][]int) bool {
	graph := make([][]int, numCourses)
	indegree := make([]int, numCourses)
	for _, p := range prerequisites {
		graph[p[1]] = append(graph[p[1]], p[0])
		indegree[p[0]]++
	}

	queue := []int{}
	for i := 0; i < numCourses; i++ {
		if indegree[i] == 0 {
			queue = append(queue, i)
		}
	}

	visited := 0
	for len(queue) > 0 {
		c := queue[0]
		queue = queue[1:]
		visited++
		for _, next := range graph[c] {
			indegree[next]--
			if indegree[next] == 0 {
				queue = append(queue, next)
			}
		}
	}
	return visited == numCourses
}

// 多源BFS：每分钟腐烂的橘子扩散到相邻
func orangesRotting(grid [][]int) int {
	m, n := len(grid), len(grid[0])
	queue := [][2]int{}
	fresh := 0

	for i := 0; i < m; i++ {
		for j := 0; j < n; j++ {
			if grid[i][j] == 2 {
				queue = append(queue, [2]int{i, j})
			} else if grid[i][j] == 1 {
				fresh++
			}
		}
	}

	dirs := [][2]int{{1, 0}, {-1, 0}, {0, 1}, {0, -1}}
	minutes := 0
	for len(queue) > 0 && fresh > 0 {
		size := len(queue)
		for i := 0; i < size; i++ {
			c := queue[0]
			queue = queue[1:]
			for _, d := range dirs {
				ni, nj := c[0]+d[0], c[1]+d[1]
				if ni < 0 || ni >= m || nj < 0 || nj >= n || grid[ni][nj] != 1 {
					continue
				}
				grid[ni][nj] = 2
				fresh--
				queue = append(queue, [2]int{ni, nj})
			}
		}
		minutes++
	}

	if fresh > 0 {
		return -1
	}
	return minutes
}

// 找出所有既能流向太平洋也能流向大西洋的点
func pacificAtlantic(heights [][]int) [][]int {
	m, n := len(heights), len(heights[0])
	pacific := make([][]bool, m)
	atlantic := make([][]bool, m)
	for i := range pacific {
		pacific[i] = make([]bool, n)
		atlantic[i] = make([]bool, n)
	}

	var dfs func(i, j int, visited [][]bool)
	dfs = func(i, j int, visited [][]bool) {
		visited[i][j] = true
		dirs := [][2]int{{1, 0}, {-1, 0}, {0, 1}, {0, -1}}
		for _, d := range dirs {
			ni, nj := i+d[0], j+d[1]
			if ni < 0 || ni >= m || nj < 0 || nj >= n || visited[ni][nj] {
				continue
			}
			if heights[ni][nj] >= heights[i][j] { // 反向：高度非递减
				dfs(ni, nj, visited)
			}
		}
	}

	for i := 0; i < m; i++ {
		dfs(i, 0, pacific)
		dfs(i, n-1, atlantic)
	}
	for j := 0; j < n; j++ {
		dfs(0, j, pacific)
		dfs(m-1, j, atlantic)
	}

	var res [][]int
	for i := 0; i < m; i++ {
		for j := 0; j < n; j++ {
			if pacific[i][j] && atlantic[i][j] {
				res = append(res, []int{i, j})
			}
		}
	}
	return res
}