From 0b0b25222b3131ada79ff78c28eb8bf07d977deb Mon Sep 17 00:00:00 2001 From: historia <[not public]> Date: Tue, 15 Sep 2026 15:35:52 -0400 Subject: feat: single-owner game state, transactional lifecycle, durable money on server stop --- internal/net/server.go | 202 +++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 180 insertions(+), 22 deletions(-) (limited to 'internal/net/server.go') diff --git a/internal/net/server.go b/internal/net/server.go index 9b5be18..b0407fc 100644 --- a/internal/net/server.go +++ b/internal/net/server.go @@ -1,6 +1,7 @@ package net import ( + "context" "crypto/rand" "crypto/rsa" "crypto/tls" @@ -73,6 +74,19 @@ type Session struct { promptVisible bool writeMu sync.Mutex + // Outbound queueing. Transport sessions (telnet/TLS/websocket) are put + // into queued mode by startWriter before their input goroutine begins: + // Write* then enqueues into outbox, and the dedicated writer goroutine + // performs the network I/O with a per-message write deadline. A stalled + // or slow client therefore blocks at most its own queue, never the + // world; an overflowing queue disconnects only that client. Sessions + // without a writer (tests, synthetic conns) keep the legacy synchronous + // path so message construction is shared and behavior identical. + outbox chan string + writerOnce sync.Once + closeOnce sync.Once + writeDone chan struct{} + // seq is a monotonically increasing sequence number assigned when the // session is added to a Hub. Hub.AllSessions and Hub.PlayersInRoom sort // sessions by seq so per-tick consumers iterate players in a stable, @@ -97,6 +111,10 @@ type Server struct { httpsTLSConfig *tls.Config hub *Hub handler func(*Session, string) + + done chan struct{} + shutdownOnce sync.Once + listenErr error } type Hub struct { @@ -107,6 +125,64 @@ type Hub struct { onDisconnect func(*Session) } +// writeQueueBounds bound the outbound queue: maxQueuedMessages before a +// client is disconnected for falling behind, writeDeadline per message, and +// closeFlushTimeout to drain the queue when the session is closing. +const ( + maxQueuedMessages = 512 + writeDeadline = 5 * time.Second + closeFlushTimeout = 2 * time.Second +) + +// startWriter puts the session into queued mode: writes are enqueued and a +// dedicated goroutine drains the queue to the network with per-message write +// deadlines. The goroutine closes the transport after the queue drains, so a +// Close() during shutdown flushes pending output before the socket drops. +// Must be called before the session's input goroutine starts. +func (sess *Session) StartWriter() { + sess.writerOnce.Do(func() { + sess.outbox = make(chan string, maxQueuedMessages) + sess.writeDone = make(chan struct{}) + go func() { + defer close(sess.writeDone) + for msg := range sess.outbox { + if dl, ok := sess.Conn.(interface { + SetWriteDeadline(time.Time) error + }); ok { + _ = dl.SetWriteDeadline(time.Now().Add(writeDeadline)) + } + if _, err := sess.Conn.Write([]byte(msg)); err != nil { + log.Printf("session write error: %v", err) + // Abandon the queue and drop the transport; the input + // goroutine notices the closed connection on its next + // read and performs normal removal. + sess.Conn.Close() + return + } + } + // Queue closed and drained: flush any final bytes and close. + if dl, ok := sess.Conn.(interface { + SetWriteDeadline(time.Time) error + }); ok { + _ = dl.SetWriteDeadline(time.Now().Add(closeFlushTimeout)) + } + sess.Conn.Close() + }() + }) +} + +// enqueue appends msg to the outbound queue. When the client's queue is full +// the session is disconnected instead of blocking the caller: the world must +// never wait on a single slow client. +func (sess *Session) enqueue(msg string) { + select { + case sess.outbox <- msg: + default: + log.Printf("session write queue overflow; disconnecting slow client") + sess.Close() + } +} + func NewHub() *Hub { return &Hub{ sessions: make(map[*Session]bool), @@ -145,24 +221,34 @@ func (h *Hub) Remove(s *Session) { } return } - h.hardRemoveLocked(s) + callback := h.hardRemoveLocked(s) h.mu.Unlock() + // The removal callback runs outside the Hub mutex: it performs saves + // and game-state cleanup, and must never serialize behind (or hold up) + // Hub membership operations. + if callback != nil { + callback(s) + } } func (h *Hub) HardRemove(s *Session) { h.mu.Lock() - defer h.mu.Unlock() - h.hardRemoveLocked(s) + callback := h.hardRemoveLocked(s) + h.mu.Unlock() + if callback != nil { + callback(s) + } } -func (h *Hub) hardRemoveLocked(s *Session) { +// hardRemoveLocked removes the session from all membership structures and +// returns the removal callback for the caller to invoke AFTER releasing the +// Hub mutex (nil when none is registered). +func (h *Hub) hardRemoveLocked(s *Session) func(*Session) { delete(h.sessions, s) for _, room := range h.rooms { delete(room, s) } - if h.onRemove != nil { - h.onRemove(s) - } + return h.onRemove } func (h *Hub) EnterRoom(s *Session, roomID int) { @@ -187,6 +273,28 @@ func (h *Hub) leaveRoomLocked(s *Session) { } } +// RekeyRooms remaps Hub room membership buckets per idMap (oldID -> newID). +// Uses delete-all-then-set-all so a swap (A<->B) cannot clobber an entry +// mid-rekey. Sessions' own player state is migrated by the game layer; this +// only moves the membership buckets. +func (h *Hub) RekeyRooms(idMap map[int]int) { + h.mu.Lock() + defer h.mu.Unlock() + if len(idMap) == 0 { + return + } + moved := make(map[int]map[*Session]bool, len(h.rooms)) + for id, set := range h.rooms { + if newID, ok := idMap[id]; ok && newID != id { + moved[newID] = set + delete(h.rooms, id) + } + } + for newID, set := range moved { + h.rooms[newID] = set + } +} + func (h *Hub) AllSessions() []*Session { h.mu.Lock() defer h.mu.Unlock() @@ -215,6 +323,7 @@ func NewServer(cfg *config.Config) (*Server, error) { s := &Server{ config: cfg, hub: NewHub(), + done: make(chan struct{}), } if cfg.Telnet.Enabled { @@ -337,7 +446,38 @@ func (s *Server) ListenAndServe(handler func(*Session, string)) error { return fmt.Errorf("no listeners configured") } - select {} + // Block until Shutdown is called (or a fatal listener error occurs). + <-s.done + return s.listenErr +} + +// Shutdown stops all transports: HTTP(S) servers drain, listeners close, and +// every session is closed (its outbound queue drains first). ListenAndServe +// returns after Shutdown completes. It is safe to call more than once. +func (s *Server) Shutdown() { + s.shutdownOnce.Do(func() { + if s.httpServer != nil { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _ = s.httpServer.Shutdown(ctx) + } + if s.httpsServer != nil { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + _ = s.httpsServer.Shutdown(ctx) + } + if s.telnetLn != nil { + s.telnetLn.Close() + } + if s.telnetTLSLn != nil { + s.telnetTLSLn.Close() + } + for _, sess := range s.hub.AllSessions() { + sess.Close() + } + close(s.done) + }) + <-s.done } func (s *Server) serveTelnet() { @@ -399,7 +539,7 @@ func (s *Server) handleSession(sess *Session, handler func(*Session, string)) { sess.Close() }() - sess.Conn.Write([]byte("\033[2J\033[H")) + sess.Write("\033[2J\033[H") sess.Write(WelcomeBanner()) sess.Write("What's your account name? ") @@ -439,15 +579,26 @@ func (sess *Session) ClearPrompt() { sess.promptVisible = false } +// dispatch delivers a fully-formatted message to the transport: enqueued in +// queued mode (the writer goroutine owns the network), written synchronously +// in direct mode (tests, synthetic conns). Caller holds writeMu. +func (sess *Session) dispatch(msg string) { + if sess.outbox != nil { + sess.enqueue(msg) + return + } + sess.Conn.Write([]byte(msg)) +} + // Write sends raw text with no wrapping. Used for prompts and partial lines. func (sess *Session) Write(msg string) { sess.writeMu.Lock() defer sess.writeMu.Unlock() if sess.promptVisible { - sess.Conn.Write([]byte("\r\n")) + sess.dispatch("\r\n") sess.promptVisible = false } - sess.Conn.Write([]byte(msg)) + sess.dispatch(msg) } func (sess *Session) WriteLine(msg string) { @@ -455,10 +606,10 @@ func (sess *Session) WriteLine(msg string) { sess.writeMu.Lock() defer sess.writeMu.Unlock() if sess.promptVisible { - sess.Conn.Write([]byte("\r\n")) + sess.dispatch("\r\n") sess.promptVisible = false } - sess.Conn.Write([]byte(msg + "\r\n")) + sess.dispatch(msg + "\r\n") } func (sess *Session) WriteLines(lines ...string) { @@ -466,10 +617,10 @@ func (sess *Session) WriteLines(lines ...string) { defer sess.writeMu.Unlock() for i, l := range lines { if i == 0 && sess.promptVisible { - sess.Conn.Write([]byte("\r\n")) + sess.dispatch("\r\n") sess.promptVisible = false } - sess.Conn.Write([]byte(sess.wrap(l) + "\r\n")) + sess.dispatch(sess.wrap(l) + "\r\n") } } @@ -478,10 +629,10 @@ func (sess *Session) Writef(format string, args ...interface{}) { sess.writeMu.Lock() defer sess.writeMu.Unlock() if sess.promptVisible { - sess.Conn.Write([]byte("\r\n")) + sess.dispatch("\r\n") sess.promptVisible = false } - sess.Conn.Write([]byte(msg)) + sess.dispatch(msg) } func (sess *Session) WritePrompt(prompt string) { @@ -490,19 +641,26 @@ func (sess *Session) WritePrompt(prompt string) { if sess.promptVisible { return } - sess.Conn.Write([]byte(prompt)) + sess.dispatch(prompt) sess.promptVisible = true } func (sess *Session) Reprompt(prompt string) { sess.writeMu.Lock() defer sess.writeMu.Unlock() - sess.Conn.Write([]byte(prompt)) + sess.dispatch(prompt) sess.promptVisible = true } +// Close stops the session. In queued mode it closes the queue so the writer +// goroutine drains pending messages and then closes the transport; in direct +// mode it closes the transport immediately. Idempotent. func (sess *Session) Close() error { - sess.writeMu.Lock() - defer sess.writeMu.Unlock() - return sess.Conn.Close() + if sess.outbox != nil { + sess.closeOnce.Do(func() { close(sess.outbox) }) + return nil + } + var err error + sess.closeOnce.Do(func() { err = sess.Conn.Close() }) + return err } -- cgit v1.2.3