Fix: player session never recovers when server resets

api/admin_test.go

@@ -174,6 +174,62 @@ func TestAdminResetPreservesLeadersAndClearsFlag(t *testing.T) {
 	}
 }
 
+func TestAdminResetPreservesConnectedSessionsAndLeaderAuthorization(t *testing.T) {
+	players := new(Players)
+	players.Init()
+	game := new(Game)
+	sockets := new(Sockets)
+	sockets.Init(players, game)
+	admin := new(Admin)
+	admin.Init(players, sockets, "top-secret", game)
+	cookie := registerTestAdmin(t, admin, "top-secret")
+
+	leaderID := players.New(TypeAntiPacLeader, "Leader", StatusDisc)
+	activeID := players.New(TypePacman, "Active", StatusDisc)
+	leaderConnection := newTestConnection(leaderID)
+	activeConnection := newTestConnection(activeID)
+	sockets.hub.registerConnection(leaderConnection)
+	sockets.hub.registerConnection(activeConnection)
+	drainTestMessages(leaderConnection)
+	drainTestMessages(activeConnection)
+
+	request := httptest.NewRequest(http.MethodPost, "/api/admin/reset", nil)
+	request.AddCookie(cookie)
+	response := httptest.NewRecorder()
+	admin.ServeHTTP(response, request)
+	if response.Code != http.StatusNoContent {
+		t.Fatalf("reset status = %d, want 204", response.Code)
+	}
+
+	if len(players.players) != 2 {
+		t.Errorf("player count after reset = %d, want 2", len(players.players))
+	}
+	if leader := players.Get(leaderID); leader == nil || leader.Type != TypeAntiPacLeader {
+		t.Errorf("leader after reset = %#v, want AntiPac Leader", leader)
+	}
+	if active := players.Get(activeID); active == nil || active.Type != TypeGhost {
+		t.Errorf("active player after reset = %#v, want Ghost", active)
+	}
+	if !sockets.hub.hasConnectionForID(leaderID) || !sockets.hub.hasConnectionForID(activeID) {
+		t.Error("admin reset replaced a connected player session")
+	}
+	if leader, _, authorized := players.LeaderState(leaderID); !authorized || leader.ID != leaderID || leader.Type != TypeAntiPacLeader {
+		t.Errorf("leader authorization after reset = %#v, authorized %v", leader, authorized)
+	}
+
+	leaderUpdate := informPlayer(t, receiveTestMessage(t, leaderConnection))
+	if leaderUpdate.ID != activeID || leaderUpdate.Type != TypeGhost {
+		t.Errorf("leader's active-player reset update = %#v, want Ghost for %q", leaderUpdate, activeID)
+	}
+	updated := informPlayer(t, receiveTestMessage(t, activeConnection))
+	if updated.ID != activeID || updated.Type != TypeGhost {
+		t.Errorf("active reset update = %#v, want Ghost for %q", updated, activeID)
+	}
+	if len(leaderConnection.send) != 0 || len(activeConnection.send) != 0 {
+		t.Errorf("unexpected extra reset messages: leader=%d active=%d", len(leaderConnection.send), len(activeConnection.send))
+	}
+}
+
 func TestAdminFlagUpdatesSharedStateAndSocketClients(t *testing.T) {
 	players := new(Players)
 	players.Init()

api/etc.go

@@ -21,6 +21,7 @@ const (
 	CMD_INFORM = "inform" // inform another player change/connection
 	CMD_REMOVE = "remove" // remove a player marker without disclosing a location
 	CMD_STATE  = "state"  // inform clients of shared game state
+	CMD_SHUTDOWN = "shutdown" // inform clients of server shutdown
 
 	// player type
 	TypeHidden        PlayerType = 0

api/hub.go

@@ -23,6 +23,12 @@ type Hub struct {
 	inform       chan PlayerID
 	state        chan GameState
 	clearOffline chan chan struct{}
+	shutdown     chan shutdownEvent
+}
+
+type shutdownEvent struct {
+	command string
+	done    chan struct{}
 }
 
 func NewHub(players *Players, games ...*Game) *Hub {
@@ -38,6 +44,7 @@ func NewHub(players *Players, games ...*Game) *Hub {
 		inform:             make(chan PlayerID),
 		state:              make(chan GameState),
 		clearOffline:       make(chan chan struct{}),
+		shutdown:           make(chan shutdownEvent),
 	}
 	if len(games) > 0 {
 		hub.game = games[0]
@@ -61,6 +68,9 @@ func (h *Hub) Run() {
 		case done := <-h.clearOffline:
 			h.clearOfflineLocations()
 			close(done)
+		case event := <-h.shutdown:
+			h.broadcastShutDown(event.command)
+			close(event.done)
 		}
 	}
 }
@@ -178,7 +188,7 @@ func (h *Hub) unregisterConnection(connection *Conn) {
 	if hasCoordinate && isMapVisibleRole(player.Type) {
 		message, ok := informMessage(player, coordinate)
 		if ok {
-			h.broadcast(message, nil, onlyViewers)
+			h.broadcastControl(message, nil, onlyViewers)
 		}
 	} else {
 		h.broadcastRemove(connection.playerID, onlyViewers, nil)
@@ -293,15 +303,46 @@ func (h *Hub) broadcast(message []byte, origin *Conn, include connectionFilter)
 	}
 }
 
+func (h *Hub) broadcastControl(message []byte, origin *Conn, include connectionFilter) {
+	var failedConnections []*Conn
+	for connection := range h.connections {
+		if connection == origin || !include(connection) {
+			continue
+		}
+		if !h.enqueueControl(connection, message) {
+			failedConnections = append(failedConnections, connection)
+		}
+	}
+	for _, connection := range failedConnections {
+		h.unregisterConnection(connection)
+	}
+}
+
 func (h *Hub) enqueue(connection *Conn, message []byte) bool {
 	select {
 	case connection.send <- message:
 		return true
 	default:
+		// Channel buffer is full. Client cannot keep up with real-time updates.
 		return false
 	}
 }
 
+func (h *Hub) enqueueControl(connection *Conn, message []byte) bool {
+	if h.enqueue(connection, message) {
+		return true
+	}
+	for {
+		select {
+		case <-connection.send:
+			// Drain older queued messages until the buffer is empty to prioritize this critical control message.
+		default:
+			// Buffer is now completely drained. Enqueue the critical control message.
+			return h.enqueue(connection, message)
+		}
+	}
+}
+
 func informMessage(player PlayerResponse, coordinate Coordinate) ([]byte, bool) {
 	playerJSON, err := json.Marshal(player)
 	if err != nil {
@@ -351,7 +392,7 @@ func (h *Hub) broadcastInform(playerID PlayerID, origin *Conn) {
 			if h.connectionCanSee(connection, playerID, player.Type) {
 				outgoing = message
 			}
-			if !h.enqueue(connection, outgoing) {
+			if !h.enqueueControl(connection, outgoing) {
 				slowConnections = append(slowConnections, connection)
 			}
 		}
@@ -367,7 +408,7 @@ func (h *Hub) broadcastInform(playerID PlayerID, origin *Conn) {
 	if retained && isMapVisibleRole(player.Type) {
 		message, ok := informMessage(player, coordinate)
 		if ok {
-			h.broadcast(message, nil, onlyViewers)
+			h.broadcastControl(message, nil, onlyViewers)
 		}
 		return
 	}
@@ -411,7 +452,7 @@ func removeMessage(playerID PlayerID) []byte {
 }
 
 func (h *Hub) broadcastRemove(playerID PlayerID, include connectionFilter, origin *Conn) {
-	h.broadcast(removeMessage(playerID), origin, include)
+	h.broadcastControl(removeMessage(playerID), origin, include)
 }
 
 func (h *Hub) clearOfflineLocations() {
@@ -420,3 +461,15 @@ func (h *Hub) clearOfflineLocations() {
 		h.broadcastRemove(playerID, onlyViewers, nil)
 	}
 }
+
+func (h *Hub) broadcastShutDown(command string) {
+	message, err := json.Marshal(Message{Command: command})
+	if err != nil {
+		return
+	}
+	for connection := range h.connections {
+		if !h.enqueueControl(connection, message) {
+			h.unregisterConnection(connection)
+		}
+	}
+}

api/socket.go

@@ -36,6 +36,13 @@ type Sockets struct {
 	hub     *Hub
 }
 
+// BroadcastShutDown sends a shutdown command to all connected clients.
+func (s *Sockets) BroadcastShutDown(command string) {
+	done := make(chan struct{})
+	s.hub.shutdown <- shutdownEvent{command: command, done: done}
+	<-done
+}
+
 func (s *Sockets) Init(players *Players, games ...*Game) {
 	s.players = players
 	s.hub = NewHub(players, games...)

api/socket_test.go

@@ -31,6 +31,112 @@ func TestPlayerStaysConnectedUntilLastSocketDisconnects(t *testing.T) {
 	}
 }
 
+func TestBroadcastShutdownQueuesShutdownCommand(t *testing.T) {
+	players := new(Players)
+	players.Init()
+	playerID := players.New(TypeGhost, "Player", StatusDisc)
+	hub := NewHub(players)
+	connection := newTestConnection(playerID)
+	hub.registerConnection(connection)
+	drainTestMessages(connection)
+
+	hub.broadcastShutDown(CMD_SHUTDOWN)
+	message := receiveTestMessage(t, connection)
+	if message.Command != CMD_SHUTDOWN {
+		t.Errorf("shutdown command = %q, want %q", message.Command, CMD_SHUTDOWN)
+	}
+}
+
+func TestBroadcastShutdownEvictsQueuedMessages(t *testing.T) {
+	players := new(Players)
+	players.Init()
+	playerID := players.New(TypeGhost, "Player", StatusDisc)
+	hub := NewHub(players)
+	connection := &Conn{
+		playerID: playerID,
+		role:     playerConnection,
+		send:     make(chan []byte, 2),
+	}
+	hub.connections[connection] = struct{}{}
+	connection.send <- []byte("stale move")
+	connection.send <- []byte("stale state")
+
+	hub.broadcastShutDown(CMD_SHUTDOWN)
+
+	if _, exists := hub.connections[connection]; !exists {
+		t.Fatal("connection was unregistered after successful prioritized delivery")
+	}
+	if len(connection.send) != 1 {
+		t.Fatalf("queued messages = %d, want 1", len(connection.send))
+	}
+	if message := receiveTestMessage(t, connection); message.Command != CMD_SHUTDOWN {
+		t.Errorf("shutdown command = %q, want %q", message.Command, CMD_SHUTDOWN)
+	}
+}
+
+func TestBroadcastShutdownUnregistersUndeliverableConnection(t *testing.T) {
+	players := new(Players)
+	players.Init()
+	playerID := players.New(TypeGhost, "Player", StatusDisc)
+	hub := NewHub(players)
+	connection := &Conn{
+		playerID: playerID,
+		role:     playerConnection,
+		send:     make(chan []byte),
+	}
+	hub.connections[connection] = struct{}{}
+
+	hub.broadcastShutDown(CMD_SHUTDOWN)
+
+	if _, exists := hub.connections[connection]; exists {
+		t.Error("undeliverable connection remains registered")
+	}
+}
+
+func TestControlBroadcastEvictsQueuedMessagesForResetUpdates(t *testing.T) {
+	players := new(Players)
+	players.Init()
+	activeID := players.New(TypePacman, "Active", StatusDisc)
+	offlineID := players.New(TypeGhost, "Offline", StatusDisc)
+	hub := NewHub(players)
+	active := &Conn{
+		playerID: activeID,
+		role:     playerConnection,
+		send:     make(chan []byte, 1),
+	}
+	viewer := &Conn{
+		role: viewerConnection,
+		send: make(chan []byte, 1),
+	}
+	hub.connections[active] = struct{}{}
+	hub.connections[viewer] = struct{}{}
+	hub.coordinates[activeID] = Coordinate{Latitude: 49.27, Longitude: -122.91}
+	hub.offlineCoordinates[offlineID] = Coordinate{Latitude: 49.28, Longitude: -122.90}
+	active.send <- []byte("stale state")
+	viewer.send <- []byte("stale move")
+
+	hub.clearOfflineLocations()
+	if _, exists := hub.connections[viewer]; !exists {
+		t.Fatal("viewer was unregistered after prioritized marker removal")
+	}
+	removed := receiveTestMessage(t, viewer)
+	if removed.Command != CMD_REMOVE || removed.Data != string(offlineID) {
+		t.Errorf("offline marker removal = %#v", removed)
+	}
+
+	if _, _, found := players.Update(activeID, TypeGhost); !found {
+		t.Fatal("reset active player update failed")
+	}
+	hub.broadcastInform(activeID, nil)
+	if _, exists := hub.connections[active]; !exists {
+		t.Fatal("active connection was unregistered after prioritized reset update")
+	}
+	updated := informPlayer(t, receiveTestMessage(t, active))
+	if updated.ID != activeID || updated.Type != TypeGhost {
+		t.Errorf("reset player update = %#v", updated)
+	}
+}
+
 func TestGameStateSnapshotAndBroadcastDoNotChangePlayerConnectionCounts(t *testing.T) {
 	players := new(Players)
 	players.Init()