From d2269ba4f9525c9785ae178e7ba1b518af5e183b Mon Sep 17 00:00:00 2001 From: Erik Dubbelboer Date: Sat, 24 Oct 2026 17:36:22 +0200 Subject: [PATCH 1/2] Fix expired peer cleanup and disconnect handling Tic Tac Toe Ultimate is generating peers quicker than our cleanup process can currently handle. The game is very popular. Clean up expired peers in batches of 250 and update each affected lobby once per batch. Add a GIN index for lobby membership lookups and elect replacement leaders once per affected lobby. Replace the startup timestamp reset with a reconnect grace period so restarts don't revive stale peers or block reconnections. Mark peers disconnected on explicit close and prevent late heartbeats from refreshing them. Cancel connections after failed pings and bound heartbeat database updates with a timeout. --- internal/signaling/handler.go | 9 +- internal/signaling/peer.go | 8 +- internal/signaling/stores/postgres.go | 132 ++++++++++--------- internal/signaling/stores/shared.go | 9 +- internal/signaling/timeout_manager.go | 63 +++++---- migrations/1792857600_cleanup_peers.down.sql | 1 + migrations/1792857600_cleanup_peers.up.sql | 1 + migrations/latest.lock | 2 +- 8 files changed, 131 insertions(+), 94 deletions(-) create mode 100644 migrations/1792857600_cleanup_peers.down.sql create mode 100644 migrations/1792857600_cleanup_peers.up.sql diff --git a/internal/signaling/handler.go b/internal/signaling/handler.go index 489f7f1..1803fee 100644 --- a/internal/signaling/handler.go +++ b/internal/signaling/handler.go @@ -104,6 +104,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 +123,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,6 +135,8 @@ 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. diff --git a/internal/signaling/peer.go b/internal/signaling/peer.go index bf039d5..332139b 100644 --- a/internal/signaling/peer.go +++ b/internal/signaling/peer.go @@ -288,7 +288,11 @@ 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 != "" { + if err := p.store.MarkPeerAsDisconnected(ctx, p.ID); err != nil { + return fmt.Errorf("unable to mark peer as disconnected: %w", err) + } + } logger.Debug("client closed", zap.String("game", p.Game), @@ -322,6 +326,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..a267eab 100644 --- a/internal/signaling/stores/postgres.go +++ b/internal/signaling/stores/postgres.go @@ -441,15 +441,16 @@ func (s *PostgresStore) UpdatePeerGeo(ctx context.Context, peerID string, countr } func (s *PostgresStore) MarkPeerAsActive(ctx context.Context, peerID 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 + WHERE peer = $2 AND disconnected = FALSE `, now, peerID) return err } @@ -516,86 +517,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..ea00902 100644 --- a/internal/signaling/stores/shared.go +++ b/internal/signaling/stores/shared.go @@ -39,8 +39,7 @@ type Store interface { 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 + 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..f575ad2 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)) } } } @@ -123,6 +130,8 @@ func (manager *TimeoutManager) Reconnected(ctx context.Context, peerID, secret, } func (manager *TimeoutManager) MarkPeerAsActive(ctx context.Context, peerID string) { + ctx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() logger := logging.GetLogger(ctx) err := manager.Store.MarkPeerAsActive(ctx, peerID) diff --git a/migrations/1792857600_cleanup_peers.down.sql b/migrations/1792857600_cleanup_peers.down.sql new file mode 100644 index 0000000..424b069 --- /dev/null +++ b/migrations/1792857600_cleanup_peers.down.sql @@ -0,0 +1 @@ +DROP INDEX CONCURRENTLY 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..8aa5740 --- /dev/null +++ b/migrations/1792857600_cleanup_peers.up.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS lobbies_peers ON lobbies USING gin (peers); 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 From 4846a382cbcd0f972489753db9fc489d916f1966 Mon Sep 17 00:00:00 2001 From: Erik Dubbelboer Date: Sat, 24 Oct 2026 17:46:42 +0200 Subject: [PATCH 2/2] Prevent stale connections from disconnecting reconnected peers Track the current connection ID for each peer and ignore heartbeats and disconnects from older connections. This prevents delayed cleanup from marking a healthy reconnected peer as disconnected. Run empty lobby cleanup every five minutes and add a partial index for empty lobbies. Include the connection ID in the existing cleanup migration and remove concurrent index creation. --- internal/signaling/handler.go | 10 +++--- internal/signaling/peer.go | 14 +++++--- internal/signaling/stores/postgres.go | 38 ++++++++++---------- internal/signaling/stores/shared.go | 8 ++--- internal/signaling/timeout_manager.go | 12 +++---- migrations/1792857600_cleanup_peers.down.sql | 4 ++- migrations/1792857600_cleanup_peers.up.sql | 4 ++- 7 files changed, 52 insertions(+), 38 deletions(-) diff --git a/internal/signaling/handler.go b/internal/signaling/handler.go index 1803fee..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, @@ -143,7 +145,7 @@ func Handler(ctx context.Context, store stores.Store, cloudflare *cloudflare.Cre 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 332139b..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) } } @@ -289,9 +290,14 @@ func (p *Peer) HandleClosePacket(ctx context.Context, packet ClosePacket) error go metrics.Record(ctx, "client", "close", p.Game, p.ID, p.Lobby) if p.ID != "" { - if err := p.store.MarkPeerAsDisconnected(ctx, p.ID); err != nil { + 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", diff --git a/internal/signaling/stores/postgres.go b/internal/signaling/stores/postgres.go index a267eab..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,7 +440,7 @@ 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) @@ -450,37 +450,39 @@ func (s *PostgresStore) MarkPeerAsActive(ctx context.Context, peerID string) err SET last_seen = $1, updated_at = $1 - WHERE peer = $2 AND disconnected = FALSE - `, 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 } diff --git a/internal/signaling/stores/shared.go b/internal/signaling/stores/shared.go index ea00902..33c752f 100644 --- a/internal/signaling/stores/shared.go +++ b/internal/signaling/stores/shared.go @@ -34,11 +34,11 @@ 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) + 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 diff --git a/internal/signaling/timeout_manager.go b/internal/signaling/timeout_manager.go index f575ad2..f49eef9 100644 --- a/internal/signaling/timeout_manager.go +++ b/internal/signaling/timeout_manager.go @@ -106,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 } @@ -122,19 +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 index 424b069..3d0e9f3 100644 --- a/migrations/1792857600_cleanup_peers.down.sql +++ b/migrations/1792857600_cleanup_peers.down.sql @@ -1 +1,3 @@ -DROP INDEX CONCURRENTLY IF EXISTS lobbies_peers; +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 index 8aa5740..3013980 100644 --- a/migrations/1792857600_cleanup_peers.up.sql +++ b/migrations/1792857600_cleanup_peers.up.sql @@ -1 +1,3 @@ -CREATE INDEX CONCURRENTLY IF NOT EXISTS lobbies_peers ON lobbies USING gin (peers); +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);