Slavasil/ruby-sea-battle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

bugfixes and refactoring in game-server

Browse Source
This commit is contained in:
Slavasil
2026-10-27 15:55:48 +03:00
parent 3629411030
commit 31ef12b5f2
9 changed files with 732 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
game-server/internal/ws/handler.go
+140 -30
View File
@@ -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)
game-server/internal/ws/handler_test.go
+129 -2
View File
@@ -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)
}
}
}