diff --git a/internal/signaling/handler.go b/internal/signaling/handler.go index 489f7f1..972348d 100644 --- a/internal/signaling/handler.go +++ b/internal/signaling/handler.go @@ -15,10 +15,11 @@ import ( "github.com/poki/netlib/internal/metrics" "github.com/poki/netlib/internal/signaling/stores" "github.com/poki/netlib/internal/util" + "github.com/rs/xid" "go.uber.org/zap" ) -const LobbyCleanInterval = 30 * time.Minute +const LobbyCleanInterval = 5 * time.Minute const LobbyCleanThreshold = 24 * time.Hour const peerPingDuration = 2 * time.Second const peerActiveUpdateInterval = 30 * time.Second @@ -95,8 +96,9 @@ func Handler(ctx context.Context, store stores.Store, cloudflare *cloudflare.Cre } peer := &Peer{ - store: store, - conn: conn, + connectionID: xid.New().String(), + store: store, + conn: conn, retrievedIDCallback: manager.Reconnected, @@ -104,6 +106,7 @@ func Handler(ctx context.Context, store stores.Store, cloudflare *cloudflare.Cre Region: region, } defer func() { + cancel() logger.Debug("peer websocket closed", zap.String("peer", peer.ID), zap.String("game", peer.Game), zap.String("origin", r.Header.Get("Origin"))) conn.Close(websocket.StatusInternalError, "unexpected closure") // nolint:errcheck @@ -122,7 +125,11 @@ func Handler(ctx context.Context, store stores.Store, cloudflare *cloudflare.Cre for { select { case <-ticker.C: - if err := peer.Send(ctx, PingPacket{Type: "ping"}); err != nil { + // Limit ping writes so a stalled connection cannot keep this goroutine alive indefinitely. + pingCtx, pingCancel := context.WithTimeout(ctx, 10*time.Second) + err := peer.Send(pingCtx, PingPacket{Type: "ping"}) + pingCancel() + if err != nil { if !util.ShouldIgnoreNetworkError(err) { if strings.Contains(err.Error(), "write: broken pipe") || strings.Contains(err.Error(), "connection reset by peer") { logger.Warn("failed to send ping packet", zap.String("peer", peer.ID), zap.Error(err)) @@ -130,13 +137,15 @@ func Handler(ctx context.Context, store stores.Store, cloudflare *cloudflare.Cre logger.Error("failed to send ping packet", zap.String("peer", peer.ID), zap.Error(err)) } } + cancel() + return } else { // If we can send a ping packet, and the peer has an ID, we update the peer as being active. // If the peer doesn't have an ID yet, it's still in the process of connecting, so we don't update it. if peer.ID != "" { now := util.NowUTC(ctx) if lastActiveUpdate.IsZero() || now.Sub(lastActiveUpdate) >= peerActiveUpdateInterval { - manager.MarkPeerAsActive(ctx, peer.ID) + manager.MarkPeerAsActive(ctx, peer.ID, peer.connectionID) lastActiveUpdate = now } } diff --git a/internal/signaling/peer.go b/internal/signaling/peer.go index bf039d5..40b5e89 100644 --- a/internal/signaling/peer.go +++ b/internal/signaling/peer.go @@ -24,8 +24,9 @@ type Peer struct { conn *websocket.Conn closedPacketReceived bool + connectionID string - retrievedIDCallback func(context.Context, string, string, string) (bool, []string, error) + retrievedIDCallback func(context.Context, string, string, string, string) (bool, []string, error) ID string Secret string @@ -197,7 +198,7 @@ func (p *Peer) HandleHelloPacket(ctx context.Context, packet HelloPacket) error if packet.ID != "" && packet.Secret != "" { logger.Debug("peer reconnecting", zap.String("game", packet.Game), zap.String("peer", packet.ID), zap.String("version", packet.Version)) var err error - hasReconnected, reconnectingLobbies, err = p.retrievedIDCallback(ctx, packet.ID, packet.Secret, packet.Game) + hasReconnected, reconnectingLobbies, err = p.retrievedIDCallback(ctx, packet.ID, packet.Secret, packet.Game, p.connectionID) if err != nil { return fmt.Errorf("unable to reconnect: %w", err) } @@ -223,7 +224,7 @@ func (p *Peer) HandleHelloPacket(ctx context.Context, packet HelloPacket) error p.Secret = util.GenerateSecret(ctx) logger.Debug("peer connecting", zap.String("game", p.Game), zap.String("peer", p.ID), zap.String("version", packet.Version)) - if err := p.store.CreatePeer(ctx, p.ID, p.Secret, p.Game); err != nil { + if err := p.store.CreatePeer(ctx, p.ID, p.Secret, p.Game, p.connectionID); err != nil { return fmt.Errorf("unable to create peer: %w", err) } } @@ -288,7 +289,16 @@ func (p *Peer) HandleClosePacket(ctx context.Context, packet ClosePacket) error logger := logging.GetLogger(ctx) go metrics.Record(ctx, "client", "close", p.Game, p.ID, p.Lobby) - p.closedPacketReceived = true + if p.ID != "" { + disconnected, err := p.store.MarkPeerAsDisconnected(ctx, p.ID, p.connectionID) + if err != nil { + return fmt.Errorf("unable to mark peer as disconnected: %w", err) + } + if !disconnected { + p.closedPacketReceived = true + return nil + } + } logger.Debug("client closed", zap.String("game", p.Game), @@ -322,6 +332,8 @@ func (p *Peer) HandleClosePacket(ctx context.Context, packet ClosePacket) error p.Lobby = "" } + // Keep deferred disconnect handling enabled if any cleanup above fails. + p.closedPacketReceived = true return nil } diff --git a/internal/signaling/stores/postgres.go b/internal/signaling/stores/postgres.go index 027ad1f..fa0669f 100644 --- a/internal/signaling/stores/postgres.go +++ b/internal/signaling/stores/postgres.go @@ -405,7 +405,7 @@ func (s *PostgresStore) ListLobbies(ctx context.Context, game string, country, r return lobbies, nil } -func (s *PostgresStore) CreatePeer(ctx context.Context, peerID, secret, gameID string) error { +func (s *PostgresStore) CreatePeer(ctx context.Context, peerID, secret, gameID, connectionID string) error { if len(peerID) > 20 { logger := logging.GetLogger(ctx) logger.Warn("peer id too long", zap.String("peerID", peerID)) @@ -414,9 +414,9 @@ func (s *PostgresStore) CreatePeer(ctx context.Context, peerID, secret, gameID s now := util.NowUTC(ctx) _, err := s.DB.Exec(ctx, ` - INSERT INTO peers (peer, secret, game, last_seen, updated_at) - VALUES ($1, $2, $3, $4, $4) - `, peerID, secret, gameID, now) + INSERT INTO peers (peer, secret, game, last_seen, updated_at, connection_id) + VALUES ($1, $2, $3, $4, $4, $5) + `, peerID, secret, gameID, now, connectionID) if err != nil { return err } @@ -440,46 +440,49 @@ func (s *PostgresStore) UpdatePeerGeo(ctx context.Context, peerID string, countr return err } -func (s *PostgresStore) MarkPeerAsActive(ctx context.Context, peerID string) error { +func (s *PostgresStore) MarkPeerAsActive(ctx context.Context, peerID, connectionID string) error { + // Only an authenticated reconnect may reactivate a disconnected peer. + // Late heartbeats must not revive it or postpone timeout cleanup. now := util.NowUTC(ctx) _, err := s.DB.Exec(ctx, ` UPDATE peers SET - disconnected = FALSE, last_seen = $1, updated_at = $1 - WHERE peer = $2 - `, now, peerID) + WHERE peer = $2 AND connection_id = $3 AND disconnected = FALSE + `, now, peerID, connectionID) return err } -func (s *PostgresStore) MarkPeerAsDisconnected(ctx context.Context, peerID string) error { +// MarkPeerAsDisconnected ignores handlers superseded by an authenticated reconnect. +func (s *PostgresStore) MarkPeerAsDisconnected(ctx context.Context, peerID, connectionID string) (bool, error) { now := util.NowUTC(ctx) - - _, err := s.DB.Exec(ctx, ` + result, err := s.DB.Exec(ctx, ` UPDATE peers - SET - disconnected = TRUE, - updated_at = $1 - WHERE peer = $2 - `, now, peerID) - return err + SET disconnected = TRUE, updated_at = $1 + WHERE peer = $2 AND connection_id = $3 + `, now, peerID, connectionID) + if err != nil { + return false, err + } + return result.RowsAffected() > 0, nil } -func (s *PostgresStore) MarkPeerAsReconnected(ctx context.Context, peerID, secret, gameID string) (bool, []string, error) { +func (s *PostgresStore) MarkPeerAsReconnected(ctx context.Context, peerID, secret, gameID, connectionID string) (bool, []string, error) { now := util.NowUTC(ctx) result, err := s.DB.Exec(ctx, ` UPDATE peers SET disconnected = FALSE, + connection_id = $5, last_seen = $1, updated_at = $1 WHERE peer = $2 AND secret = $3 AND game = $4 - `, now, peerID, secret, gameID) + `, now, peerID, secret, gameID, connectionID) if err != nil { return false, nil, err } @@ -516,86 +519,93 @@ func (s *PostgresStore) MarkPeerAsReconnected(ctx context.Context, peerID, secre return true, lobbies, nil } -func (s *PostgresStore) ClaimNextTimedOutPeer(ctx context.Context, threshold time.Duration) (string, bool, map[string][]string, error) { +func (s *PostgresStore) ClaimTimedOutPeers(ctx context.Context, threshold time.Duration, limit int) ([]TimedOutPeer, error) { + if limit <= 0 { + return nil, fmt.Errorf("cleanup limit must be positive") + } + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() now := util.NowUTC(ctx) - tx, err := s.DB.Begin(ctx) if err != nil { - return "", false, nil, err + return nil, err } defer tx.Rollback(context.Background()) //nolint:errcheck - var peerID string - var disconnected bool - err = tx.QueryRow(ctx, ` - WITH d AS ( - SELECT peer, disconnected - FROM peers + rows, err := tx.Query(ctx, ` + WITH expired AS ( + SELECT peer FROM peers WHERE last_seen < $1 - LIMIT 1 + ORDER BY last_seen, peer + LIMIT $2 + FOR UPDATE SKIP LOCKED ) - DELETE FROM peers - USING d - WHERE peers.peer = d.peer - RETURNING d.peer, d.disconnected - `, now.Add(-threshold)).Scan(&peerID, &disconnected) + DELETE FROM peers p USING expired e + WHERE p.peer = e.peer + RETURNING p.peer + `, now.Add(-threshold), limit) if err != nil { - if errors.Is(err, pgx.ErrNoRows) { - return "", false, nil, nil + return nil, err + } + var result []TimedOutPeer + var ids []string + indexes := make(map[string]int) + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return nil, err } - return "", false, nil, err + indexes[id] = len(result) + ids = append(ids, id) + result = append(result, TimedOutPeer{ID: id, GameLobbies: make(map[string][]string)}) + } + if err := rows.Err(); err != nil { + return nil, err + } + if len(result) == 0 { + return nil, nil } - gameLobbies := make(map[string][]string) - - rows, err := tx.Query(ctx, ` - UPDATE lobbies - SET - peers = array_remove(peers, $1), + // Lock each affected lobby once, retaining its old membership for notifications. + // Array overlap uses lobbies_peers; ANY(peers) cannot use that GIN index. + rows, err = tx.Query(ctx, ` + WITH affected AS MATERIALIZED ( + SELECT game, code, peers FROM lobbies + WHERE peers && $1::varchar(20)[] + ORDER BY game, code + FOR UPDATE + ) + UPDATE lobbies l + SET peers = ARRAY(SELECT p FROM unnest(l.peers) AS p WHERE p <> ALL($1::varchar(20)[])), updated_at = $2 - WHERE $1 = ANY(peers) - RETURNING game, code - `, peerID, now) + FROM affected a + WHERE l.game = a.game AND l.code = a.code + RETURNING a.game, a.code, a.peers + `, ids, now) if err != nil { - return "", false, nil, err + return nil, err } - for rows.Next() { - var game string - var lobby string - - err = rows.Scan(&game, &lobby) - if err != nil { - return "", false, nil, err + var game, code string + var members []string + if err := rows.Scan(&game, &code, &members); err != nil { + rows.Close() + return nil, err + } + for _, id := range members { + if i, ok := indexes[id]; ok { + result[i].GameLobbies[game] = append(result[i].GameLobbies[game], code) + } } - - gameLobbies[game] = append(gameLobbies[game], lobby) } - - if err = rows.Err(); err != nil { - return "", false, nil, err + if err := rows.Err(); err != nil { + return nil, err } - - if err = tx.Commit(ctx); err != nil { - return "", false, nil, err + if err := tx.Commit(ctx); err != nil { + return nil, err } - - return peerID, disconnected, gameLobbies, nil -} - -// ResetAllPeerLastSeen will reset all last_seen. -// This is being called when the process restarts so it doesn't matter -// how long the process was down. -func (s *PostgresStore) ResetAllPeerLastSeen(ctx context.Context) error { - now := util.NowUTC(ctx) - - _, err := s.DB.Exec(ctx, ` - UPDATE peers - SET - last_seen = $1, - updated_at = $1 - `, now) - return err + return result, nil } func (s *PostgresStore) CleanEmptyLobbies(ctx context.Context, olderThan time.Time) error { diff --git a/internal/signaling/stores/shared.go b/internal/signaling/stores/shared.go index ae466e4..33c752f 100644 --- a/internal/signaling/stores/shared.go +++ b/internal/signaling/stores/shared.go @@ -34,13 +34,12 @@ type Store interface { Subscribe(ctx context.Context, callback SubscriptionCallback, game, lobby, peerID string) Publish(ctx context.Context, topic string, data []byte) error - CreatePeer(ctx context.Context, peerID, secret, gameID string) error + CreatePeer(ctx context.Context, peerID, secret, gameID, connectionID string) error UpdatePeerGeo(ctx context.Context, peerID string, country, region string) error - MarkPeerAsActive(ctx context.Context, peerID string) error - MarkPeerAsDisconnected(ctx context.Context, peerID string) error - MarkPeerAsReconnected(ctx context.Context, peerID, secret, gameID string) (bool, []string, error) - ClaimNextTimedOutPeer(ctx context.Context, threshold time.Duration) (string, bool, map[string][]string, error) - ResetAllPeerLastSeen(ctx context.Context) error + MarkPeerAsActive(ctx context.Context, peerID, connectionID string) error + MarkPeerAsDisconnected(ctx context.Context, peerID, connectionID string) (bool, error) + MarkPeerAsReconnected(ctx context.Context, peerID, secret, gameID, connectionID string) (bool, []string, error) + ClaimTimedOutPeers(ctx context.Context, threshold time.Duration, limit int) ([]TimedOutPeer, error) CleanEmptyLobbies(ctx context.Context, olderThan time.Time) error @@ -83,3 +82,9 @@ type ElectionResult struct { Leader string Term int } + +// TimedOutPeer includes the lobby memberships removed in the same transaction. +type TimedOutPeer struct { + ID string + GameLobbies map[string][]string +} diff --git a/internal/signaling/timeout_manager.go b/internal/signaling/timeout_manager.go index 3b04708..f49eef9 100644 --- a/internal/signaling/timeout_manager.go +++ b/internal/signaling/timeout_manager.go @@ -19,21 +19,27 @@ type TimeoutManager struct { } func (manager *TimeoutManager) Run(ctx context.Context) { - logger := logging.GetLogger(ctx) - - if err := manager.Store.ResetAllPeerLastSeen(ctx); err != nil { - logger.Error("failed to reset all peer last seen", zap.Error(err)) - } - if manager.DisconnectThreshold == 0 { // We update peer activity every 30 seconds. Make it possible to somehow miss // two updates before we consider a peer timed out. manager.DisconnectThreshold = time.Second * 90 } + // Allow clients to reconnect after a restart without refreshing stale rows. + grace := time.NewTimer(manager.DisconnectThreshold) + defer grace.Stop() + select { + case <-ctx.Done(): + return + case <-grace.C: + } for ctx.Err() == nil { - time.Sleep(time.Second) manager.RunOnce(ctx) + select { + case <-ctx.Done(): + return + case <-time.After(time.Second): + } } } @@ -41,31 +47,32 @@ func (manager *TimeoutManager) RunOnce(ctx context.Context) { logger := logging.GetLogger(ctx) for ctx.Err() == nil { - peerID, disconnected, gameLobbies, err := manager.Store.ClaimNextTimedOutPeer(ctx, manager.DisconnectThreshold) + peers, err := manager.Store.ClaimTimedOutPeers(ctx, manager.DisconnectThreshold, 250) if err != nil { - logger.Error("failed to claim next timedout peer", zap.Error(err)) + logger.Error("failed to claim timedout peers", zap.Error(err)) + return } - if peerID == "" { - break + if len(peers) == 0 { + return } - - for gameID, lobbies := range gameLobbies { - for _, lobbyCode := range lobbies { - logger.Info("peer timeout", zap.String("peer", peerID), zap.String("game", gameID), zap.String("lobby", lobbyCode)) - - if err := manager.disconnectPeerInLobby(ctx, peerID, gameID, lobbyCode); err != nil { - logger.Error("failed to disconnect peer", zap.Error(err), zap.String("peer", peerID), zap.String("game", gameID), zap.String("lobby", lobbyCode)) - } - - // If the peer wasn't disconnected normally, they might still be the leader of a lobby. - // Just to be sure, do a leader election. - if !disconnected { - if err := manager.doLeaderElectionAndPublish(ctx, gameID, lobbyCode); err != nil { - logger.Error("failed to do leader election", zap.Error(err), zap.String("peer", peerID), zap.String("game", gameID), zap.String("lobby", lobbyCode)) + // Several expired peers can share a lobby. Elect once after all removals. + type lobbyKey struct{ game, code string } + lobbies := make(map[lobbyKey]struct{}) + for _, peer := range peers { + for game, codes := range peer.GameLobbies { + for _, code := range codes { + logger.Info("peer timeout", zap.String("peer", peer.ID), zap.String("game", game), zap.String("lobby", code)) + if err := manager.disconnectPeerInLobby(ctx, peer.ID, game, code); err != nil { + logger.Error("failed to disconnect peer", zap.Error(err), zap.String("peer", peer.ID)) } + lobbies[lobbyKey{game, code}] = struct{}{} + go metrics.Record(ctx, "client", "timeout", game, peer.ID, code) } - - go metrics.Record(ctx, "client", "timeout", gameID, peerID, lobbyCode) + } + } + for lobby := range lobbies { + if err := manager.doLeaderElectionAndPublish(ctx, lobby.game, lobby.code); err != nil { + logger.Error("failed to do leader election", zap.Error(err), zap.String("game", lobby.game), zap.String("lobby", lobby.code)) } } } @@ -99,13 +106,13 @@ func (manager *TimeoutManager) Disconnected(ctx context.Context, p *Peer) { } logger.Debug("peer marked as disconnected", zap.String("id", p.ID), zap.String("lobby", p.Lobby)) - err := manager.Store.MarkPeerAsDisconnected(ctx, p.ID) + disconnected, err := manager.Store.MarkPeerAsDisconnected(ctx, p.ID, p.connectionID) if err != nil { logger.Error("failed to record timeout peer", zap.Error(err)) return } - if p.Lobby == "" { + if !disconnected || p.Lobby == "" { return } @@ -115,17 +122,19 @@ func (manager *TimeoutManager) Disconnected(ctx context.Context, p *Peer) { } } -func (manager *TimeoutManager) Reconnected(ctx context.Context, peerID, secret, gameID string) (bool, []string, error) { +func (manager *TimeoutManager) Reconnected(ctx context.Context, peerID, secret, gameID, connectionID string) (bool, []string, error) { logger := logging.GetLogger(ctx) logger.Debug("peer marked as reconnected", zap.String("peer", peerID)) - return manager.Store.MarkPeerAsReconnected(ctx, peerID, secret, gameID) + return manager.Store.MarkPeerAsReconnected(ctx, peerID, secret, gameID, connectionID) } -func (manager *TimeoutManager) MarkPeerAsActive(ctx context.Context, peerID string) { +func (manager *TimeoutManager) MarkPeerAsActive(ctx context.Context, peerID, connectionID string) { + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() logger := logging.GetLogger(ctx) - err := manager.Store.MarkPeerAsActive(ctx, peerID) + err := manager.Store.MarkPeerAsActive(ctx, peerID, connectionID) // context.Canceled is expected when the connection is closed right as this function // is called. We don't want to log this as an error. diff --git a/migrations/1792857600_cleanup_peers.down.sql b/migrations/1792857600_cleanup_peers.down.sql new file mode 100644 index 0000000..3d0e9f3 --- /dev/null +++ b/migrations/1792857600_cleanup_peers.down.sql @@ -0,0 +1,3 @@ +ALTER TABLE peers DROP COLUMN connection_id; +DROP INDEX IF EXISTS lobbies_empty_updated_at; +DROP INDEX IF EXISTS lobbies_peers; diff --git a/migrations/1792857600_cleanup_peers.up.sql b/migrations/1792857600_cleanup_peers.up.sql new file mode 100644 index 0000000..3013980 --- /dev/null +++ b/migrations/1792857600_cleanup_peers.up.sql @@ -0,0 +1,3 @@ +CREATE INDEX IF NOT EXISTS lobbies_peers ON lobbies USING gin (peers); +CREATE INDEX IF NOT EXISTS lobbies_empty_updated_at ON lobbies (updated_at) WHERE peers = '{}'; +ALTER TABLE peers ADD COLUMN connection_id VARCHAR(20); diff --git a/migrations/latest.lock b/migrations/latest.lock index d372081..a6a12a7 100644 --- a/migrations/latest.lock +++ b/migrations/latest.lock @@ -1 +1 @@ -1770020948_latency3 +1792857600_cleanup_peers