diff --git a/game-server/README.md b/game-server/README.md index 21c6d96..6c0abb7 100644 --- a/game-server/README.md +++ b/game-server/README.md @@ -53,10 +53,10 @@ JWT_SECRET=... PUBLIC_ADDR=ws://localhost:8081/ws ./server Проверки при JOIN: -- подпись (HS256) и срок действия обоих токенов; +- подпись (HS256) и срок действия обоих токенов (при первом подключении к партии); - `game_server` строго равен `PUBLIC_ADDR` этого сервера; -- `player1_id != player2_id`; -- `user_id` из пользовательского токена совпадает с `player1_id` или +- `player1_id > 0` и `player2_id > 0`, а также `player1_id != player2_id`; +- `user_id > 0` из пользовательского токена совпадает с `player1_id` или `player2_id` игрового токена; - партия не находится в реестре завершённых (см. ниже). @@ -108,19 +108,24 @@ JWT_SECRET=... PUBLIC_ADDR=ws://localhost:8081/ws ./server | Код | Значение | |------|----------| | 1000 | нормальное завершение (партия окончена, END уже отправлен) | +| 1001 | сервер завершает работу (graceful shutdown) | | 4000 | соединение заменено новым соединением того же игрока | | 4001 | первое сообщение не JOIN | | 4002 | сообщение не соответствует формату | -| 4003 | недействительный игровой токен / чужой `game_server` | -| 4004 | недействительный пользовательский токен / игрок не участник партии | +| 4003 | недействительный игровой токен / чужой `game_server` / неположительные id | +| 4004 | недействительный пользовательский токен / игрок не участник партии / неположительный id | ## Обработка краевых случаев - **Переподключение.** Игрок идентифицируется токенами, а не соединением. - Новое JOIN от того же игрока закрывает старое соединение (код 4000) и - занимает его слот; очередь неподтверждённых сообщений переживает - разрыв и переотправляется. Одновременно существует не более одного - соединения на игрока. + Игровой токен запоминается в структуре матча при первом подключении: + повторное подключение с тем же игровым токеном полностью пропускает + его валидацию (включая проверку срока действия). Новое JOIN от того же + игрока закрывает старое соединение (код 4000) и занимает его слот; + очередь неподтверждённых сообщений переживает разрыв и переотправляется. + Одновременно существует не более одного соединения на игрока. +- **Повторный JOIN.** Если клиент присылает повторный JOIN на уже + установленном соединении, такое сообщение просто игнорируется без ошибки. - **Повторный JOIN после завершения партии.** Завершённые партии хранятся в реестре (`match_id` -> END-сообщения обоих игроков) до истечения срока действия игрового токена. Поздний JOIN получает в ответ сохранённое END @@ -131,8 +136,13 @@ JWT_SECRET=... PUBLIC_ADDR=ws://localhost:8081/ws ./server токена: после истечения токена второй игрок всё равно не сможет войти. - **JOIN с токеном, указывающим на другой сервер** (неверный `game_server`) отклоняется с кодом 4003. -- **`player1_id == player2_id`** отклоняется (код 4003) — иначе один игрок - мог бы занять оба слота. +- **`player1_id == player2_id`** или неположительные ID отклоняются (код 4003). +- **Валидация ACK.** Поле `seq` от клиента проверяется на положительность и + ненахождение за пределами последнего отправленного `seq`. Некорректный `seq` + отклоняется с ошибкой. +- **Асинхронная отправка.** Запись сообщений в WebSocket делегирована + отдельной per-connection горутине, что предотвращает блокировку мьютексов + менеджера во время write timeout. - Неизвестные опкоды после JOIN игнорируются (совместимость с будущими версиями протокола); мусорные/не-JSON сообщения логируются и пропускаются. - Keepalive: сервер пингует соединение каждые 25 с; размер входящего @@ -166,5 +176,4 @@ go test -race ./... - завершение партии по итогам игры с персонифицированными кодами победы/поражения; - репорт итогов партии на main-server (реестр завершённых партий уже - гарантирует, что партия завершается ровно один раз); -- очистка состояния начавшихся партий после их завершения. + гарантирует, что партия завершается ровно один раз). diff --git a/game-server/cmd/server/main.go b/game-server/cmd/server/main.go index 397929b..866b81c 100644 --- a/game-server/cmd/server/main.go +++ b/game-server/cmd/server/main.go @@ -59,6 +59,7 @@ func main() { <-ctx.Done() log.Info("shutting down") + manager.Shutdown() shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _ = srv.Shutdown(shutdownCtx) diff --git a/game-server/internal/auth/jwt.go b/game-server/internal/auth/jwt.go index 3efc3f7..864499e 100644 --- a/game-server/internal/auth/jwt.go +++ b/game-server/internal/auth/jwt.go @@ -16,6 +16,7 @@ type GameClaims struct { Player1ID int64 `json:"player1_id"` Player2ID int64 `json:"player2_id"` GameServer string `json:"game_server"` + RawToken string `json:"-"` jwt.RegisteredClaims } @@ -48,6 +49,28 @@ func ParseGameToken(secret []byte, tokenString string) (*GameClaims, error) { if claims.MatchID == "" { return nil, fmt.Errorf("invalid game token: empty match_id") } + if claims.Player1ID <= 0 || claims.Player2ID <= 0 { + return nil, fmt.Errorf("invalid game token: player IDs must be positive") + } + claims.RawToken = tokenString + return claims, nil +} + +// ParseGameTokenUnverified extracts claims from a game token without verifying +// its signature or expiration. +func ParseGameTokenUnverified(tokenString string) (*GameClaims, error) { + claims := &GameClaims{} + _, _, err := jwt.NewParser().ParseUnverified(tokenString, claims) + if err != nil { + return nil, fmt.Errorf("invalid game token: %w", err) + } + if claims.MatchID == "" { + return nil, fmt.Errorf("invalid game token: empty match_id") + } + if claims.Player1ID <= 0 || claims.Player2ID <= 0 { + return nil, fmt.Errorf("invalid game token: player IDs must be positive") + } + claims.RawToken = tokenString return claims, nil } @@ -62,5 +85,8 @@ func ParseUserToken(secret []byte, tokenString string) (*UserClaims, error) { if err != nil { return nil, fmt.Errorf("invalid user token: %w", err) } + if claims.UserID <= 0 { + return nil, fmt.Errorf("invalid user token: user_id must be positive") + } return claims, nil } diff --git a/game-server/internal/auth/jwt_test.go b/game-server/internal/auth/jwt_test.go index 86f439d..c0f1675 100644 --- a/game-server/internal/auth/jwt_test.go +++ b/game-server/internal/auth/jwt_test.go @@ -63,6 +63,37 @@ func TestParseGameTokenMissingExp(t *testing.T) { } } +func TestParseGameTokenNonPositiveIDs(t *testing.T) { + for _, tc := range [][2]int64{{0, 2}, {-1, 2}, {1, 0}, {1, -5}} { + token := sign(t, &GameClaims{ + MatchID: "m1", + Player1ID: tc[0], + Player2ID: tc[1], + RegisteredClaims: jwt.RegisteredClaims{ + ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Hour)), + }, + }, secret) + if _, err := ParseGameToken(secret, token); err == nil { + t.Fatalf("expected error for non-positive IDs: %+v", tc) + } + } +} + +func TestParseGameTokenUnverified(t *testing.T) { + // Works even if expired + token := gameToken(t, secret, time.Now().Add(-time.Hour)) + claims, err := ParseGameTokenUnverified(token) + if err != nil { + t.Fatalf("parse unverified: %v", err) + } + if claims.MatchID != "m1" || claims.Player1ID != 1 || claims.Player2ID != 2 { + t.Fatalf("unexpected claims: %+v", claims) + } + if claims.RawToken != token { + t.Fatalf("raw token mismatch: got %q, want %q", claims.RawToken, token) + } +} + func TestParseUserToken(t *testing.T) { token := sign(t, &UserClaims{ UserID: 42, @@ -76,3 +107,15 @@ func TestParseUserToken(t *testing.T) { t.Fatalf("user_id = %d, want 42", claims.UserID) } } + +func TestParseUserTokenNonPositiveID(t *testing.T) { + for _, id := range []int64{0, -1, -99} { + token := sign(t, &UserClaims{ + UserID: id, + RegisteredClaims: jwt.RegisteredClaims{ExpiresAt: jwt.NewNumericDate(time.Now().Add(time.Hour))}, + }, secret) + if _, err := ParseUserToken(secret, token); err == nil { + t.Fatalf("expected error for user_id = %d", id) + } + } +} diff --git a/game-server/internal/game/manager.go b/game-server/internal/game/manager.go index 03f192a..a423476 100644 --- a/game-server/internal/game/manager.go +++ b/game-server/internal/game/manager.go @@ -16,7 +16,37 @@ import ( // ErrMatchFinished is returned by Join when the player tries to join a match // that has already ended. The stored END message (if any) is replayed to the // connection before Join returns. -var ErrMatchFinished = errors.New("match already finished") +var ( + ErrMatchFinished = errors.New("match already finished") + ErrInvalidSeq = errors.New("invalid seq") +) + +// Server-side match states. +const ( + StatePlacingShips = 1 + StatePlayer1Ready = 2 + StatePlayer2Ready = 3 + StatePlayer1Turn = 4 + StatePlayer2Turn = 5 +) + +// playerStatesFor maps a server-side state to the respective states seen by player 1 and player 2. +func playerStatesFor(serverState int) (p1State, p2State int) { + switch serverState { + case StatePlacingShips: + return protocol.StatePlacingShips, protocol.StatePlacingShips + case StatePlayer1Ready: + return protocol.StateWaitOpponentReady, protocol.StateOpponentReady + case StatePlayer2Ready: + return protocol.StateOpponentReady, protocol.StateWaitOpponentReady + case StatePlayer1Turn: + return protocol.StateYourTurn, protocol.StateWaitOpponentTurn + case StatePlayer2Turn: + return protocol.StateWaitOpponentTurn, protocol.StateYourTurn + default: + return 0, 0 + } +} // Sender abstracts a player connection so the game logic stays testable and // independent of the websocket implementation. @@ -49,6 +79,7 @@ type PlayerSlot struct { // Match is the state of a single game of sea battle. type Match struct { ID string + gameToken string // game token used when match was created tokenExp time.Time // game token expiration; bounds the join deadline playerIDs [2]int64 // slot 0 = player1, slot 1 = player2 (internal only) slots [2]*PlayerSlot @@ -56,10 +87,24 @@ type Match struct { mu sync.Mutex started bool // both players joined at least once ended bool - state int // current global game state + state int // current global server-side game state joinTimer *time.Timer } +func (m *Match) GameToken() string { + return m.gameToken +} + +func (m *Match) PlayerIDs() [2]int64 { + return m.playerIDs +} + +func (m *Match) State() int { + m.mu.Lock() + defer m.mu.Unlock() + return m.state +} + func (m *Match) slotIndex(userID int64) int { switch userID { case m.playerIDs[0]: @@ -77,6 +122,7 @@ func (m *Match) slotIndex(userID int64) int { type finishedMatch struct { ends map[int64][]byte // userID -> END message sent to that player expiresAt time.Time // game token expiration time + gameToken string } // Manager owns all matches of this server instance. @@ -96,6 +142,13 @@ func NewManager(joinTimeout time.Duration) *Manager { } } +// GetMatch returns the match with the given ID, or nil if not found. +func (m *Manager) GetMatch(matchID string) *Match { + m.mu.Lock() + defer m.mu.Unlock() + return m.matches[matchID] +} + // Join attaches conn to the match identified by the game token claims. // userID must already be validated against the claims by the caller. // @@ -120,8 +173,13 @@ func (m *Manager) Join(ctx context.Context, conn Sender, gc *auth.GameClaims, us } match, ok := m.matches[gc.MatchID] if !ok { + token := gc.RawToken + if token == "" { + token = gc.MatchID + } match = &Match{ ID: gc.MatchID, + gameToken: token, tokenExp: gc.ExpiresAt.Time, playerIDs: [2]int64{gc.Player1ID, gc.Player2ID}, slots: [2]*PlayerSlot{ @@ -133,7 +191,7 @@ func (m *Manager) Join(ctx context.Context, conn Sender, gc *auth.GameClaims, us // never later than the game token itself expires — after that a // join is impossible anyway. deadline := now.Add(m.joinTimeout) - if match.tokenExp.Before(deadline) { + if !match.tokenExp.IsZero() && match.tokenExp.Before(deadline) { deadline = match.tokenExp } match.joinTimer = time.AfterFunc(time.Until(deadline), func() { @@ -141,13 +199,29 @@ func (m *Manager) Join(ctx context.Context, conn Sender, gc *auth.GameClaims, us }) m.matches[gc.MatchID] = match } - m.mu.Unlock() + // Acquire match.mu before releasing m.mu to eliminate the race with onJoinTimeout / EndMatch match.mu.Lock() + m.mu.Unlock() defer match.mu.Unlock() if match.ended { - // Lost race with the join timeout; the match is being torn down. + // The match has ended; make sure the END message is sent before closing. + var endData []byte + idx := match.slotIndex(userID) + if idx >= 0 && len(match.slots[idx].unacked) > 0 { + endData = match.slots[idx].unacked[len(match.slots[idx].unacked)-1].data + } + if endData == nil { + m.mu.Lock() + if fin := m.finished[gc.MatchID]; fin != nil { + endData = fin.ends[userID] + } + m.mu.Unlock() + } + if endData != nil { + _ = conn.Send(ctx, endData) + } _ = conn.Close(protocol.CloseCodeNormal, "match already finished") return nil, -1, ErrMatchFinished } @@ -174,8 +248,7 @@ func (m *Manager) Join(ctx context.Context, conn Sender, gc *auth.GameClaims, us if match.joinTimer != nil { match.joinTimer.Stop() } - match.state = protocol.StatePlacingShips - m.setStateLocked(ctx, match, match.state) + m.setStateLocked(ctx, match, StatePlacingShips) case match.started: // Reconnect of a player while the game is running. m.replayLocked(ctx, slot) @@ -198,10 +271,16 @@ func (m *Manager) Leave(conn Sender, match *Match, idx int) { // Ack confirms delivery of all messages with seq <= the given sequence // number for the given player. -func (m *Manager) Ack(match *Match, idx int, seq int64) { +func (m *Manager) Ack(match *Match, idx int, seq int64) error { match.mu.Lock() defer match.mu.Unlock() + if idx < 0 || idx >= len(match.slots) { + return errors.New("invalid slot index") + } slot := match.slots[idx] + if seq <= 0 || seq > slot.seq { + return ErrInvalidSeq + } kept := slot.unacked[:0] for _, qm := range slot.unacked { if qm.seq > seq { @@ -209,13 +288,26 @@ func (m *Manager) Ack(match *Match, idx int, seq int64) { } } slot.unacked = kept + return nil } -// setStateLocked moves the match into a new state and notifies both players. -// match.mu must be held. +// SetState moves the match into a new server-side state and notifies both players. +func (m *Manager) SetState(ctx context.Context, match *Match, state int) { + match.mu.Lock() + defer match.mu.Unlock() + m.setStateLocked(ctx, match, state) +} + +// setStateLocked moves the match into a new server-side state and notifies both players +// with appropriate player-side states. match.mu must be held. func (m *Manager) setStateLocked(ctx context.Context, match *Match, state int) { - for _, slot := range match.slots { - m.queueStateLocked(ctx, slot, state) + match.state = state + p1State, p2State := playerStatesFor(state) + if p1State != 0 { + m.queueStateLocked(ctx, match.slots[0], p1State) + } + if p2State != 0 { + m.queueStateLocked(ctx, match.slots[1], p2State) } } @@ -248,6 +340,23 @@ func (m *Manager) replayLocked(ctx context.Context, slot *PlayerSlot) { } } +// EndMatch finishes a match with the given code and reason for both players. +func (m *Manager) EndMatch(matchID string, code int, reason string) { + m.mu.Lock() + match, ok := m.matches[matchID] + if !ok { + m.mu.Unlock() + return + } + match.mu.Lock() + if match.ended { + match.mu.Unlock() + m.mu.Unlock() + return + } + m.finishMatchLocked(match, code, reason) +} + // onJoinTimeout finishes a match whose second player never joined. func (m *Manager) onJoinTimeout(matchID string) { m.mu.Lock() @@ -262,8 +371,18 @@ func (m *Manager) onJoinTimeout(matchID string) { m.mu.Unlock() return } + m.finishMatchLocked(match, protocol.EndCodeJoinTimeout, "join timeout exceeded") +} + +// finishMatchLocked tears down a match, removes it from m.matches, records it in m.finished, +// and sends END notifications to connected players. Both m.mu and match.mu must be held +// upon entry; both are unlocked by this function. +func (m *Manager) finishMatchLocked(match *Match, code int, reason string) { match.ended = true - delete(m.matches, matchID) + if match.joinTimer != nil { + match.joinTimer.Stop() + } + delete(m.matches, match.ID) ends := make(map[int64][]byte, 2) type notification struct { @@ -273,7 +392,7 @@ func (m *Manager) onJoinTimeout(matchID string) { var notify []notification for _, slot := range match.slots { slot.seq++ - data := protocol.BuildEnd(slot.seq, protocol.EndCodeJoinTimeout, "join timeout exceeded") + data := protocol.BuildEnd(slot.seq, code, reason) slot.unacked = append(slot.unacked, queuedMessage{seq: slot.seq, data: data}) ends[slot.UserID] = data if slot.conn != nil { @@ -283,11 +402,19 @@ func (m *Manager) onJoinTimeout(matchID string) { } match.mu.Unlock() - // Remember the outcome until the game token expires so that late join - // attempts are answered with the same END message instead of creating a - // phantom match. - m.finished[matchID] = &finishedMatch{ends: ends, expiresAt: match.tokenExp} - m.purgeFinishedLocked(time.Now()) + // Remember outcome in finished registry. Bound expiresAt to at least now + 5m if tokenExp is in the past. + now := time.Now() + expiresAt := match.tokenExp + if expiresAt.Before(now.Add(5 * time.Minute)) { + expiresAt = now.Add(5 * time.Minute) + } + + m.finished[match.ID] = &finishedMatch{ + ends: ends, + expiresAt: expiresAt, + gameToken: match.gameToken, + } + m.purgeFinishedLocked(now) m.mu.Unlock() // Network I/O happens outside of any lock. @@ -297,6 +424,37 @@ func (m *Manager) onJoinTimeout(matchID string) { } } +// Shutdown closes all active player connections across all matches with code 1001 (going away) +// and stops all pending join timers. +func (m *Manager) Shutdown() { + m.mu.Lock() + var conns []Sender + for _, match := range m.matches { + match.mu.Lock() + if match.joinTimer != nil { + match.joinTimer.Stop() + } + for _, slot := range match.slots { + if slot.conn != nil { + conns = append(conns, slot.conn) + slot.conn = nil + } + } + match.mu.Unlock() + } + m.mu.Unlock() + + var wg sync.WaitGroup + for _, conn := range conns { + wg.Add(1) + go func(c Sender) { + defer wg.Done() + _ = c.Close(protocol.CloseCodeGoingAway, "going away") + }(conn) + } + wg.Wait() +} + // purgeFinishedLocked drops registry entries whose game token has expired. // Called lazily whenever the registry is touched. m.mu must be held. func (m *Manager) purgeFinishedLocked(now time.Time) { diff --git a/game-server/internal/game/manager_test.go b/game-server/internal/game/manager_test.go index 4a553e6..694932e 100644 --- a/game-server/internal/game/manager_test.go +++ b/game-server/internal/game/manager_test.go @@ -237,7 +237,9 @@ func TestAckThenReconnectResendsCurrentState(t *testing.T) { // Player 1 acknowledges the STATE message, then reconnects. The queue // is empty, so the server re-sends the current state with a fresh seq. - m.Ack(match, 0, 1) + if err := m.Ack(match, 0, 1); err != nil { + t.Fatalf("ack: %v", err) + } fresh := &fakeConn{} if _, _, err := m.Join(context.Background(), fresh, gc, gc.Player1ID); err != nil { @@ -253,6 +255,195 @@ func TestAckThenReconnectResendsCurrentState(t *testing.T) { } } +func TestServerStateTransitions(t *testing.T) { + m := NewManager(time.Minute) + gc := claims(time.Now().Add(time.Hour)) + c1, c2 := &fakeConn{}, &fakeConn{} + + match, _, err := m.Join(context.Background(), c1, gc, gc.Player1ID) + if err != nil { + t.Fatalf("join p1: %v", err) + } + if _, _, err := m.Join(context.Background(), c2, gc, gc.Player2ID); err != nil { + t.Fatalf("join p2: %v", err) + } + // Initial state is StatePlacingShips + if match.State() != StatePlacingShips { + t.Fatalf("state = %d, want StatePlacingShips", match.State()) + } + + // Move to StatePlayer1Ready + m.SetState(context.Background(), match, StatePlayer1Ready) + p1Msgs, p2Msgs := c1.messages(), c2.messages() + if v := parseMsg(t, p1Msgs[len(p1Msgs)-1]); v[2] != float64(protocol.StateWaitOpponentReady) { + t.Fatalf("p1 got %v, want StateWaitOpponentReady (3)", v) + } + if v := parseMsg(t, p2Msgs[len(p2Msgs)-1]); v[2] != float64(protocol.StateOpponentReady) { + t.Fatalf("p2 got %v, want StateOpponentReady (2)", v) + } + + // Move to StatePlayer1Turn + m.SetState(context.Background(), match, StatePlayer1Turn) + p1Msgs, p2Msgs = c1.messages(), c2.messages() + if v := parseMsg(t, p1Msgs[len(p1Msgs)-1]); v[2] != float64(protocol.StateYourTurn) { + t.Fatalf("p1 got %v, want StateYourTurn (4)", v) + } + if v := parseMsg(t, p2Msgs[len(p2Msgs)-1]); v[2] != float64(protocol.StateWaitOpponentTurn) { + t.Fatalf("p2 got %v, want StateWaitOpponentTurn (5)", v) + } + + // Move to StatePlayer2Turn + m.SetState(context.Background(), match, StatePlayer2Turn) + p1Msgs, p2Msgs = c1.messages(), c2.messages() + if v := parseMsg(t, p1Msgs[len(p1Msgs)-1]); v[2] != float64(protocol.StateWaitOpponentTurn) { + t.Fatalf("p1 got %v, want StateWaitOpponentTurn (5)", v) + } + if v := parseMsg(t, p2Msgs[len(p2Msgs)-1]); v[2] != float64(protocol.StateYourTurn) { + t.Fatalf("p2 got %v, want StateYourTurn (4)", v) + } +} + +func TestAckValidation(t *testing.T) { + m := NewManager(time.Minute) + gc := claims(time.Now().Add(time.Hour)) + c1, c2 := &fakeConn{}, &fakeConn{} + + match, _, err := m.Join(context.Background(), c1, gc, gc.Player1ID) + if err != nil { + t.Fatalf("join p1: %v", err) + } + if _, _, err := m.Join(context.Background(), c2, gc, gc.Player2ID); err != nil { + t.Fatalf("join p2: %v", err) + } + + // seq <= 0 should fail + if err := m.Ack(match, 0, 0); err != ErrInvalidSeq { + t.Fatalf("ack 0: got err = %v, want ErrInvalidSeq", err) + } + if err := m.Ack(match, 0, -1); err != ErrInvalidSeq { + t.Fatalf("ack -1: got err = %v, want ErrInvalidSeq", err) + } + + // seq > slot.seq (e.g. 2 when only 1 sent) should fail + if err := m.Ack(match, 0, 2); err != ErrInvalidSeq { + t.Fatalf("ack 2: got err = %v, want ErrInvalidSeq", err) + } + + // seq == 1 should succeed + if err := m.Ack(match, 0, 1); err != nil { + t.Fatalf("ack 1: unexpected err = %v", err) + } + + // Idempotent duplicate ack should succeed + if err := m.Ack(match, 0, 1); err != nil { + t.Fatalf("repeat ack 1: unexpected err = %v", err) + } +} + +func TestEndMatchCleansUpMatch(t *testing.T) { + m := NewManager(time.Minute) + gc := claims(time.Now().Add(time.Hour)) + c1, c2 := &fakeConn{}, &fakeConn{} + + match, _, err := m.Join(context.Background(), c1, gc, gc.Player1ID) + if err != nil { + t.Fatalf("join p1: %v", err) + } + if _, _, err := m.Join(context.Background(), c2, gc, gc.Player2ID); err != nil { + t.Fatalf("join p2: %v", err) + } + + // Match started successfully and is in m.matches + if m.GetMatch(gc.MatchID) == nil { + t.Fatal("expected match in m.matches") + } + + // End match + m.EndMatch(match.ID, 0, "game completed") + + // Match should be deleted from m.matches (preventing memory leak) + if m.GetMatch(gc.MatchID) != nil { + t.Fatal("expected match to be deleted from m.matches") + } + + // Both connections should be notified and closed with CloseCodeNormal + for i, c := range []*fakeConn{c1, c2} { + msgs := c.messages() + lastMsg := parseMsg(t, msgs[len(msgs)-1]) + if lastMsg[0] != float64(protocol.OpEnd) || lastMsg[3] != "game completed" { + t.Fatalf("player %d unexpected END: %v", i, lastMsg) + } + if closed, code := c.isClosed(); !closed || code != protocol.CloseCodeNormal { + t.Fatalf("player %d closed=%v code=%d, want closed with %d", i, closed, code, protocol.CloseCodeNormal) + } + } + + // Late join should replay END + c3 := &fakeConn{} + _, _, err = m.Join(context.Background(), c3, gc, gc.Player1ID) + if err != ErrMatchFinished { + t.Fatalf("err = %v, want ErrMatchFinished", err) + } + msgs := c3.messages() + if len(msgs) != 1 { + t.Fatalf("replayed %d messages, want 1", len(msgs)) + } +} + +func TestManagerShutdown(t *testing.T) { + m := NewManager(time.Minute) + gc := claims(time.Now().Add(time.Hour)) + c1, c2 := &fakeConn{}, &fakeConn{} + + _, _, err := m.Join(context.Background(), c1, gc, gc.Player1ID) + if err != nil { + t.Fatalf("join p1: %v", err) + } + if _, _, err := m.Join(context.Background(), c2, gc, gc.Player2ID); err != nil { + t.Fatalf("join p2: %v", err) + } + + m.Shutdown() + + for i, c := range []*fakeConn{c1, c2} { + if closed, code := c.isClosed(); !closed || code != protocol.CloseCodeGoingAway { + t.Fatalf("player %d closed=%v code=%d, want closed with %d", i, closed, code, protocol.CloseCodeGoingAway) + } + } +} + +func TestJoinWhenMatchAlreadyEndedReplaysReason(t *testing.T) { + m := NewManager(time.Minute) + gc := claims(time.Now().Add(time.Hour)) + c1 := &fakeConn{} + + _, _, err := m.Join(context.Background(), c1, gc, gc.Player1ID) + if err != nil { + t.Fatalf("join: %v", err) + } + + // Trigger onJoinTimeout so match is ended + m.onJoinTimeout(gc.MatchID) + + // Late join on ended match must send the reason before closing + c2 := &fakeConn{} + _, _, err = m.Join(context.Background(), c2, gc, gc.Player2ID) + if err != ErrMatchFinished { + t.Fatalf("got err = %v, want ErrMatchFinished", err) + } + msgs := c2.messages() + if len(msgs) != 1 { + t.Fatalf("got %d messages, want 1", len(msgs)) + } + v := parseMsg(t, msgs[0]) + if v[0] != float64(protocol.OpEnd) || v[2] != float64(protocol.EndCodeJoinTimeout) || v[3] != "join timeout exceeded" { + t.Fatalf("unexpected message: %v", v) + } + if closed, code := c2.isClosed(); !closed || code != protocol.CloseCodeNormal { + t.Fatalf("c2 closed=%v code=%d, want %d", closed, code, protocol.CloseCodeNormal) + } +} + func mustMatch(t *testing.T, m *Manager, id string) *Match { t.Helper() m.mu.Lock() diff --git a/game-server/internal/protocol/protocol.go b/game-server/internal/protocol/protocol.go index 1a2b05b..942741b 100644 --- a/game-server/internal/protocol/protocol.go +++ b/game-server/internal/protocol/protocol.go @@ -51,9 +51,10 @@ const ( EndCodeJoinTimeout = -1 // the second player never joined ) -// WebSocket close codes used by the server (application range 4000+). +// WebSocket close codes used by the server (application range 4000+ and RFC 6455 codes). const ( CloseCodeNormal = 1000 // normal closure + CloseCodeGoingAway = 1001 // server shutting down CloseCodeReplaced = 4000 // connection replaced by a newer one of the same player CloseCodeExpectedJoin = 4001 // first message was not a JOIN CloseCodeBadMessage = 4002 // malformed message diff --git a/game-server/internal/ws/handler.go b/game-server/internal/ws/handler.go index d5f5971..ce84181 100644 --- a/game-server/internal/ws/handler.go +++ b/game-server/internal/ws/handler.go @@ -4,9 +4,9 @@ package ws import ( "context" + "errors" "log/slog" "net/http" - "sync" "time" "github.com/coder/websocket" @@ -24,22 +24,109 @@ const ( pingTimeout = 10 * time.Second ) -// Conn adapts a websocket connection to the game.Sender interface. -type Conn struct { - ws *websocket.Conn - wmu sync.Mutex // serializes writes +type writeOp struct { + data []byte + isClose bool + code websocket.StatusCode + reason string } -func (c *Conn) Send(_ context.Context, data []byte) error { - ctx, cancel := context.WithTimeout(context.Background(), writeTimeout) - defer cancel() - c.wmu.Lock() - defer c.wmu.Unlock() - return c.ws.Write(ctx, websocket.MessageText, data) +// Conn adapts a websocket connection to the game.Sender interface, +// delegating all writes to a dedicated goroutine to prevent locking +// game mutexes during write timeouts. +type Conn struct { + ws *websocket.Conn + sendCh chan writeOp + done chan struct{} // closed to abort writeLoop + closed chan struct{} // closed when writeLoop exits +} + +func newConn(ws *websocket.Conn) *Conn { + c := &Conn{ + ws: ws, + sendCh: make(chan writeOp, 64), + done: make(chan struct{}), + closed: make(chan struct{}), + } + go c.writeLoop() + return c +} + +func (c *Conn) writeLoop() { + defer close(c.closed) + + for { + select { + case <-c.done: + return + case op, ok := <-c.sendCh: + if !ok { + return + } + if op.isClose { + _ = c.ws.Close(op.code, op.reason) + return + } + ctx, cancel := context.WithTimeout(context.Background(), writeTimeout) + err := c.ws.Write(ctx, websocket.MessageText, op.data) + cancel() + if err != nil { + _ = c.ws.Close(websocket.StatusGoingAway, "write error") + return + } + } + } +} + +func (c *Conn) Send(ctx context.Context, data []byte) error { + select { + case <-c.done: + return errors.New("connection closed") + default: + } + + msg := make([]byte, len(data)) + copy(msg, data) + + select { + case <-c.done: + return errors.New("connection closed") + case <-ctx.Done(): + return ctx.Err() + case c.sendCh <- writeOp{data: msg}: + return nil + } } func (c *Conn) Close(code int, reason string) error { - return c.ws.Close(websocket.StatusCode(code), reason) + select { + case <-c.closed: + return nil + default: + } + + op := writeOp{ + isClose: true, + code: websocket.StatusCode(code), + reason: reason, + } + + select { + case c.sendCh <- op: + select { + case <-c.closed: + case <-time.After(writeTimeout): + _ = c.ws.Close(websocket.StatusCode(code), reason) + } + default: + select { + case <-c.done: + default: + close(c.done) + } + _ = c.ws.Close(websocket.StatusCode(code), reason) + } + return nil } // Handler is the http.Handler of the websocket endpoint. @@ -104,22 +191,40 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { return } - // Validate the game token and make sure the token was issued for this - // particular server instance (strict comparison with the public address). - gc, err := auth.ParseGameToken(h.JWTSecret, gameToken) - if err != nil { - h.log.Debug("bad game token", "err", err) - wsc.Close(protocol.CloseCodeBadGameToken, "invalid game token") - return + // If player is reconnecting with the same game token used when creating the match, + // skip token validation completely. + var gc *auth.GameClaims + unverified, _ := auth.ParseGameTokenUnverified(gameToken) + if unverified != nil { + if existing := h.Manager.GetMatch(unverified.MatchID); existing != nil && existing.GameToken() == gameToken { + gc = &auth.GameClaims{ + MatchID: existing.ID, + Player1ID: existing.PlayerIDs()[0], + Player2ID: existing.PlayerIDs()[1], + RawToken: gameToken, + } + } } - if gc.GameServer != h.PublicAddr { - h.log.Debug("game token issued for another server", "want", h.PublicAddr, "got", gc.GameServer) - wsc.Close(protocol.CloseCodeBadGameToken, "game server mismatch") - return - } - if gc.Player1ID == gc.Player2ID { - wsc.Close(protocol.CloseCodeBadGameToken, "invalid match") - return + + if gc == nil { + // Validate the game token and make sure the token was issued for this + // particular server instance (strict comparison with the public address). + var err error + gc, err = auth.ParseGameToken(h.JWTSecret, gameToken) + if err != nil { + h.log.Debug("bad game token", "err", err) + wsc.Close(protocol.CloseCodeBadGameToken, "invalid game token") + return + } + if gc.GameServer != h.PublicAddr { + h.log.Debug("game token issued for another server", "want", h.PublicAddr, "got", gc.GameServer) + wsc.Close(protocol.CloseCodeBadGameToken, "game server mismatch") + return + } + if gc.Player1ID <= 0 || gc.Player2ID <= 0 || gc.Player1ID == gc.Player2ID { + wsc.Close(protocol.CloseCodeBadGameToken, "invalid match") + return + } } // Validate the user token and make sure the user is a participant. @@ -129,12 +234,12 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { wsc.Close(protocol.CloseCodeBadUserToken, "invalid user token") return } - if uc.UserID != gc.Player1ID && uc.UserID != gc.Player2ID { + if uc.UserID <= 0 || (uc.UserID != gc.Player1ID && uc.UserID != gc.Player2ID) { wsc.Close(protocol.CloseCodeBadUserToken, "not a participant of this match") return } - conn := &Conn{ws: wsc} + conn := newConn(wsc) match, slotIdx, err := h.Manager.Join(ctx, conn, gc, uc.UserID) if err != nil { // ErrMatchFinished: the END message has already been replayed and @@ -160,13 +265,18 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { continue } switch opcode { + case protocol.OpJoin: + // Duplicate JOIN on an established connection is ignored. case protocol.OpAck: seq, err := protocol.ParseAck(args) if err != nil { h.log.Debug("malformed ACK", "match", match.ID, "user", uc.UserID, "err", err) continue } - h.Manager.Ack(match, slotIdx, seq) + if err := h.Manager.Ack(match, slotIdx, seq); err != nil { + h.log.Debug("invalid ACK seq", "match", match.ID, "user", uc.UserID, "seq", seq, "err", err) + continue + } default: // Unknown opcodes are ignored for forward compatibility. h.log.Debug("unknown opcode", "match", match.ID, "user", uc.UserID, "opcode", opcode) diff --git a/game-server/internal/ws/handler_test.go b/game-server/internal/ws/handler_test.go index 06a943a..3519c71 100644 --- a/game-server/internal/ws/handler_test.go +++ b/game-server/internal/ws/handler_test.go @@ -52,14 +52,19 @@ func userToken(t *testing.T, userID int64) string { return token } -func startServer(t *testing.T, joinTimeout time.Duration) (url string) { +func startServerWithManager(t *testing.T, joinTimeout time.Duration) (string, *game.Manager) { t.Helper() manager := game.NewManager(joinTimeout) log := slog.New(slog.NewTextHandler(testWriter{t}, nil)) handler := ws.NewHandler(manager, secret, publicAddr, []string{"*"}, log) srv := httptest.NewServer(handler) t.Cleanup(srv.Close) - return "ws" + strings.TrimPrefix(srv.URL, "http") + return "ws" + strings.TrimPrefix(srv.URL, "http"), manager +} + +func startServer(t *testing.T, joinTimeout time.Duration) (url string) { + u, _ := startServerWithManager(t, joinTimeout) + return u } type testWriter struct{ t *testing.T } @@ -190,3 +195,125 @@ func TestReconnectDuringGame(t *testing.T) { t.Fatalf("unexpected replayed message: %v", v) } } + +func TestReconnectWithExpiredTokenAllowedIfSameToken(t *testing.T) { + url := startServer(t, time.Minute) + // Short expiration: 1200ms + exp := time.Now().Add(1200 * time.Millisecond) + gt := gameToken(t, "m-reconnect-exp", 1, 2, publicAddr, exp) + + c1, c2 := dial(t, url), dial(t, url) + join(t, c1, gt, userToken(t, 1)) + join(t, c2, gt, userToken(t, 2)) + readMsg(t, c1) // STATE + readMsg(t, c2) // STATE + + // Wait for the game token to expire + time.Sleep(1300 * time.Millisecond) + + // Simulate disconnect + c1.Close(websocket.StatusAbnormalClosure, "network gone") + + // Reconnect with the SAME token (which is now expired) + c1b := dial(t, url) + join(t, c1b, gt, userToken(t, 1)) + v := readMsg(t, c1b) + if v[0] != float64(3) || v[1] != float64(1) || v[2] != float64(1) { + t.Fatalf("unexpected replayed message on reconnect: %v", v) + } + + // But an expired token for a non-existent match is rejected with 4003 + gtExpired := gameToken(t, "m-other", 1, 2, publicAddr, time.Now().Add(-time.Hour)) + cOther := dial(t, url) + join(t, cOther, gtExpired, userToken(t, 1)) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, _, err := cOther.Read(ctx) + if err == nil { + t.Fatal("expected connection to be rejected") + } + if code := websocket.CloseStatus(err); code != 4003 { + t.Fatalf("close code = %d, want 4003", code) + } +} + +func TestDuplicateJoinIgnored(t *testing.T) { + url := startServer(t, time.Minute) + exp := time.Now().Add(time.Hour) + gt := gameToken(t, "m-dup-join", 1, 2, publicAddr, exp) + + c1, c2 := dial(t, url), dial(t, url) + join(t, c1, gt, userToken(t, 1)) + join(t, c2, gt, userToken(t, 2)) + readMsg(t, c1) // STATE + + // Send duplicate JOIN on c1 + join(t, c1, gt, userToken(t, 1)) + + // Send ACK after duplicate JOIN; connection must still be open and functioning + ackMsg, _ := json.Marshal([]any{4, 1}) + if err := c1.Write(context.Background(), websocket.MessageText, ackMsg); err != nil { + t.Fatalf("write ACK: %v", err) + } +} + +func TestNonPositiveIDsRejected(t *testing.T) { + url := startServer(t, time.Minute) + exp := time.Now().Add(time.Hour) + + // Non-positive Player1ID + gtBadP1 := gameToken(t, "m-bad-p1", 0, 2, publicAddr, exp) + c := dial(t, url) + join(t, c, gtBadP1, userToken(t, 2)) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, _, err := c.Read(ctx) + if err == nil || websocket.CloseStatus(err) != 4003 { + t.Fatalf("expected 4003 for player1_id = 0, got err = %v", err) + } + + // Non-positive UserID + gtValid := gameToken(t, "m-valid", 1, 2, publicAddr, exp) + c2 := dial(t, url) + join(t, c2, gtValid, userToken(t, -1)) + ctx2, cancel2 := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel2() + _, _, err = c2.Read(ctx2) + if err == nil || websocket.CloseStatus(err) != 4004 { + t.Fatalf("expected 4004 for user_id = -1, got err = %v", err) + } +} + +func TestManagerShutdownClosesConnectionsGoingAway(t *testing.T) { + url, mgr := startServerWithManager(t, time.Minute) + exp := time.Now().Add(time.Hour) + gt := gameToken(t, "m-shutdown", 1, 2, publicAddr, exp) + + c1, c2 := dial(t, url), dial(t, url) + join(t, c1, gt, userToken(t, 1)) + join(t, c2, gt, userToken(t, 2)) + readMsg(t, c1) + readMsg(t, c2) + + errCh := make(chan error, 2) + for _, c := range []*websocket.Conn{c1, c2} { + go func(conn *websocket.Conn) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _, _, err := conn.Read(ctx) + errCh <- err + }(c) + } + + mgr.Shutdown() + + for i := 0; i < 2; i++ { + err := <-errCh + if err == nil { + t.Fatalf("client %d: expected connection to be closed", i) + } + if code := websocket.CloseStatus(err); code != 1001 { + t.Fatalf("client %d: close status = %d, want 1001", i, code) + } + } +}