《游戏服务端架构·品类、状态与拓扑》5.3 会话、重连、顶号与连接迁移

CONNECTED --authenticate--> AUTHENTICATED --bind--> BOUND
BOUND --drain--> DRAINING --flush--> RESUMABLE --resume--> BOUND
任意非 CLOSED 状态 --close--> CLOSED

if session.state != BOUND: reject
if message.epoch != session.epoch: reject
accept

// file: internal/session/registry.go
// tier: L1
// verify: go test ./internal/session/
package session

import (
	"fmt"
	"sync"
	"time"
)

// Registry is an in-memory directory. A production implementation stores the
// same fields in a shared store with compare-and-set on Epoch.
type Registry struct {
	mu        sync.Mutex
	window    time.Duration
	now       func() time.Time
	seq       uint64
	sessions  map[string]*Session
	bound     map[string]string
	lastEpoch map[string]uint64
	replay    map[string]Session
}

func NewRegistryWithClock(resumeWindow time.Duration, now func() time.Time) *Registry {
	if now == nil {
		now = time.Now
	}
	return &Registry{
		window: resumeWindow, now: now,
		sessions: make(map[string]*Session), bound: make(map[string]string),
		lastEpoch: make(map[string]uint64), replay: make(map[string]Session),
	}
}

// Connect allocates an unauthenticated session for a gateway connection.
func (r *Registry) Connect(gatewayID string) (Session, error) {
	if gatewayID == "" {
		return Session{}, fmt.Errorf("%w: empty gateway", ErrBadState)
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	r.seq++
	s := &Session{ID: fmt.Sprintf("s-%d", r.seq), GatewayID: gatewayID, State: Connected}
	r.sessions[s.ID] = s
	return *s, nil
}

// Authenticate attaches an account but does not yet take over the account.
func (r *Registry) Authenticate(id, accountID string) (Session, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[id]
	if !ok {
		return Session{}, ErrNotFound
	}
	if accountID == "" || !CanTransition(s.State, Authenticated) {
		return Session{}, ErrBadState
	}
	s.AccountID = accountID
	s.State = Authenticated
	return *s, nil
}

// Bind completes login. If the account already has a session, that session is
// closed as kicked and the new binding gets a higher epoch.
func (r *Registry) Bind(id string) (Session, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[id]
	if !ok {
		return Session{}, ErrNotFound
	}
	if !CanTransition(s.State, Bound) {
		return Session{}, ErrBadState
	}
	if oldID := r.bound[s.AccountID]; oldID != "" && oldID != id {
		if old := r.sessions[oldID]; old != nil {
			r.closeLocked(old, ReasonKicked)
		}
	}
	s.Epoch = r.nextEpoch(s.AccountID)
	s.State = Bound
	s.ResumeUntil = time.Time{}
	r.bound[s.AccountID] = id
	return *s, nil
}

// Drain stops writes but keeps the binding for a possible resume.
func (r *Registry) Drain(id string) (Session, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[id]
	if !ok {
		return Session{}, ErrNotFound
	}
	if !CanTransition(s.State, Draining) {
		return Session{}, ErrBadState
	}
	s.State = Draining
	s.ResumeUntil = r.now().Add(r.window)
	return *s, nil
}

// MarkResumable is called after the transport has flushed its last message.
func (r *Registry) MarkResumable(id string) (Session, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[id]
	if !ok {
		return Session{}, ErrNotFound
	}
	if !CanTransition(s.State, Resumable) {
		return Session{}, ErrBadState
	}
	if r.now().After(s.ResumeUntil) {
		r.closeLocked(s, ReasonExpired)
		return Session{}, ErrResumeExpired
	}
	s.State = Resumable
	return *s, nil
}

// Resume rebinds a resumable session. requestID makes a retried resume return
// the same epoch instead of bumping it twice.
func (r *Registry) Resume(id string, epoch uint64, gatewayID, requestID string) (Session, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[id]
	if !ok {
		return Session{}, ErrNotFound
	}
	if requestID != "" {
		if prev, done := r.replay[id+"|"+requestID]; done {
			if prev.Epoch != s.Epoch {
				return Session{}, ErrStaleEpoch
			}
			if s.State != Bound {
				return Session{}, ErrBadState
			}
			return *s, nil
		}
	}
	if epoch != s.Epoch {
		return Session{}, ErrStaleEpoch
	}
	if !CanTransition(s.State, Bound) {
		return Session{}, ErrBadState
	}
	if r.now().After(s.ResumeUntil) {
		r.closeLocked(s, ReasonExpired)
		return Session{}, ErrResumeExpired
	}
	if gatewayID == "" {
		return Session{}, ErrBadState
	}
	s.Epoch = r.nextEpoch(s.AccountID)
	s.GatewayID = gatewayID
	s.State = Bound
	s.ResumeUntil = time.Time{}
	r.bound[s.AccountID] = id
	if requestID != "" {
		r.replay[id+"|"+requestID] = *s
	}
	return *s, nil
}

// Migrate moves a live binding to another gateway and increments the epoch.
// The old connection's next write fails with ErrStaleEpoch.
func (r *Registry) Migrate(id, targetGateway string) (Session, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[id]
	if !ok {
		return Session{}, ErrNotFound
	}
	if s.State != Bound {
		return Session{}, ErrBadState
	}
	if targetGateway == "" {
		return Session{}, ErrBadState
	}
	if targetGateway == s.GatewayID {
		return Session{}, ErrSameGateway
	}
	s.Epoch = r.nextEpoch(s.AccountID)
	s.GatewayID = targetGateway
	return *s, nil
}

// AuthorizeWrite is the fence every business message passes through.
func (r *Registry) AuthorizeWrite(id string, epoch uint64) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[id]
	if !ok {
		return ErrNotFound
	}
	if s.State != Bound {
		return ErrNotWritable
	}
	if epoch != s.Epoch {
		return ErrStaleEpoch
	}
	return nil
}

// Close is idempotent.
func (r *Registry) Close(id string, reason EndReason) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[id]
	if !ok {
		return ErrNotFound
	}
	r.closeLocked(s, reason)
	return nil
}

func (r *Registry) Get(id string) (Session, bool) {
	r.mu.Lock()
	defer r.mu.Unlock()
	s, ok := r.sessions[id]
	if !ok {
		return Session{}, false
	}
	return *s, true
}

func (r *Registry) nextEpoch(accountID string) uint64 {
	r.lastEpoch[accountID]++
	return r.lastEpoch[accountID]
}

func (r *Registry) closeLocked(s *Session, reason EndReason) {
	if s.State == Closed {
		return
	}
	s.State = Closed
	s.EndReason = reason
	s.ResumeUntil = time.Time{}
	if r.bound[s.AccountID] == s.ID {
		delete(r.bound, s.AccountID)
	}
}

$ cd .local/checks/gsarch-lab
$ gofmt -l internal/session
$ go vet ./internal/session/
$ go test -count=1 ./internal/session/ -v
...
PASS
ok  	gsarchlab/internal/session	0.241s