// internal/kvstore/store.go
type Store interface {
    Range(key, rangeEnd string, limit int64, revision int64) (*RangeResponse, error)
    PutWithLease(key, value string, leaseID int64) (int64, *KeyValue, error)
    DeleteRange(key, rangeEnd string) (int64, []*KeyValue, int64, error)
    // ...
}

// api/etcd/kv.go
resp, err := s.server.store.Range(key, rangeEnd, limit, revision)
if err != nil {
    return nil, toGRPCError(err)  // ✅ 不忽略错误
}

// api/etcd/auth_manager.go
if err := json.Marshal(user); err != nil {
    return fmt.Errorf("failed to marshal user: %w", err)  // ✅ 使用 %w 包装
}

// api/etcd/watch_manager.go
type WatchManager struct {
    mu       sync.RWMutex
    nextID   atomic.Int64    // ✅ 使用 atomic
    stopped  atomic.Bool     // ✅ 使用 atomic.Bool
}

func (wm *WatchManager) Create(...) {
    wm.mu.Lock()
    defer wm.mu.Unlock()  // ✅ 使用 defer
    // ...
}

// ✅ 好的命名
type WatchManager struct { }  // Manager 后缀
type LeaseManager struct { }
type Store interface { }      // 接口名词

func (wm *WatchManager) Create(...) int64 { }  // 方法动词开头
func (lm *LeaseManager) Grant(...) { }

// WatchManager 管理所有的 watch 订阅
type WatchManager struct { }

// Create 创建一个新的 watch
func (wm *WatchManager) Create(...) int64 { }

// api/etcd/watch_manager.go:132
func (wm *WatchManager) Stop() {
    if !wm.stopped.CompareAndSwap(false, true) {
        return  // ✅ 防止重复关闭
    }

    wm.mu.Lock()
    defer wm.mu.Unlock()  // ✅ 使用 defer

    for watchID := range wm.watches {
        wm.store.CancelWatch(watchID)  // ✅ 清理资源
    }
    wm.watches = make(map[int64]*watchStream)  // ✅ 重置状态
}

import (
    pb "go.etcd.io/etcd/api/v3/etcdserverpb"
    mvccpb "go.etcd.io/etcd/api/v3/mvccpb"
)

// ✅ 完全兼容 etcd client
pb.RegisterKVServer(grpcSrv, &KVServer{server: s})
pb.RegisterWatchServer(grpcSrv, &WatchServer{server: s})

// api/etcd/server.go:151-156
pb.RegisterKVServer(grpcSrv, &KVServer{server: s})
pb.RegisterWatchServer(grpcSrv, &WatchServer{server: s})
pb.RegisterLeaseServer(grpcSrv, &LeaseServer{server: s})
pb.RegisterMaintenanceServer(grpcSrv, &MaintenanceServer{server: s})
pb.RegisterAuthServer(grpcSrv, &AuthServer{server: s})

// api/etcd/server.go:303
func (s *Server) getResponseHeader() *pb.ResponseHeader {
    return &pb.ResponseHeader{
        ClusterId: s.clusterID,
        MemberId:  s.memberID,
        Revision:  s.store.CurrentRevision(),
        RaftTerm:  s.store.GetRaftStatus().Term,  // ✅ 使用真实 Raft 状态
    }
}

// api/etcd/errors.go
func toGRPCError(err error) error {
    switch {
    case errors.Is(err, ErrKeyNotFound):
        return status.Error(codes.NotFound, err.Error())  // ✅ 使用标准错误码
    case errors.Is(err, ErrCompacted):
        return status.Error(codes.OutOfRange, err.Error())
    case errors.Is(err, ErrLeaseNotFound):
        return status.Error(codes.NotFound, err.Error())
    default:
        return status.Error(codes.Internal, err.Error())
    }
}

// api/etcd/kv.go:137
func (s *KVServer) Txn(ctx context.Context, req *pb.TxnRequest) (*pb.TxnResponse, error) {
    // ✅ Compare-Then-Else 语义
    cmps := make([]kvstore.Compare, len(req.Compare))
    thenOps := make([]kvstore.Op, len(req.Success))
    elseOps := make([]kvstore.Op, len(req.Failure))

    // ✅ 原子性保证
    txnResp, err := s.server.store.Txn(cmps, thenOps, elseOps)
    // ...
}

// pkg/log/logger.go
log.Info("Starting etcd-compatible gRPC server",
    log.String("address", s.listener.Addr().String()),
    log.Component("server"))

log.Error("Failed to revoke expired lease",
    log.Int64("lease_id", id),
    log.Error("error", err))

// api/etcd/server.go:185-244
shutdownMgr.RegisterHook(reliability.PhaseStopAccepting, ...)
shutdownMgr.RegisterHook(reliability.PhaseDrainConnections, ...)
shutdownMgr.RegisterHook(reliability.PhasePersistState, ...)
shutdownMgr.RegisterHook(reliability.PhaseCloseResources, ...)

// api/etcd/server.go:312
func (s *Server) PanicRecoveryInterceptor(...) {
    defer func() {
        if r := recover(); r != nil {
            reliability.RecoverPanic(...)  // ✅ 恢复并记录
            err = fmt.Errorf("internal server error: panic recovered")
        }
    }()
    return handler(ctx, req)
}

// api/etcd/server.go:160-182
healthpb.RegisterHealthServer(grpcSrv, healthMgr.GetServer())

healthMgr.RegisterChecker(reliability.NewStorageHealthChecker("storage", ...))
healthMgr.RegisterChecker(reliability.NewLeaseHealthChecker("lease", ...))

// 缺失的指标
- etcd_server_requests_total              // 请求总数
- etcd_server_request_duration_seconds    // 请求延迟
- etcd_server_failed_requests_total       // 失败请求数
- etcd_mvcc_db_total_size_in_bytes        // 数据库大小
- etcd_network_client_grpc_received_bytes_total  // 网络流量

import "go.opentelemetry.io/otel"

func (s *KVServer) Put(ctx context.Context, req *pb.PutRequest) (*pb.PutResponse, error) {
    ctx, span := otel.Tracer("etcdapi").Start(ctx, "KV.Put")
    defer span.End()

    span.SetAttributes(
        attribute.String("key", string(req.Key)),
        attribute.Int64("lease", req.Lease),
    )
    // ...
}

// api/etcd/auth_manager.go:59
func (am *AuthManager) loadState() error {
    resp, err := am.store.Range(authEnabledKey, "", 1, 0)  // ❌ 无 context
    // ...
}

func (am *AuthManager) loadState(ctx context.Context) error {
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
    defer cancel()

    resp, err := am.store.RangeWithContext(ctx, authEnabledKey, "", 1, 0)
    // ...
}

func (s *Server) RequestIDInterceptor(...) {
    requestID := uuid.New().String()
    ctx = context.WithValue(ctx, "request_id", requestID)

    log.Info("Request received",
        log.String("request_id", requestID),
        log.String("method", info.FullMethod))

    return handler(ctx, req)
}

import "golang.org/x/time/rate"

type RateLimiter struct {
    limiters sync.Map  // clientIP -> *rate.Limiter
}

func (rl *RateLimiter) Allow(clientIP string) bool {
    limiter, _ := rl.limiters.LoadOrStore(clientIP,
        rate.NewLimiter(rate.Limit(100), 200))  // 100 QPS, burst 200
    return limiter.(*rate.Limiter).Allow()
}

func (am *AuthManager) AddUser(name, password string) error {
    // ...

    // 审计日志
    audit.Log(audit.UserAdded, map[string]interface{}{
        "username":  name,
        "admin":     getCurrentUser(ctx),
        "timestamp": time.Now(),
        "client_ip": getClientIP(ctx),
    })

    return nil
}