(c *connIO)
| 1001 | } |
| 1002 | |
| 1003 | func (q *querier) cleanupConn(c *connIO) { |
| 1004 | logger := q.logger.With(slog.F("peer_id", c.UniqueID())) |
| 1005 | q.mu.Lock() |
| 1006 | defer q.mu.Unlock() |
| 1007 | |
| 1008 | mk := mKey(c.UniqueID()) |
| 1009 | mpr, ok := q.mappers[mk] |
| 1010 | if !ok { |
| 1011 | return |
| 1012 | } |
| 1013 | if mpr.c != c { |
| 1014 | logger.Debug(q.ctx, "attempt to cleanup for duplicate connection, ignoring") |
| 1015 | return |
| 1016 | } |
| 1017 | err := c.CoordinatorClose() |
| 1018 | if err != nil { |
| 1019 | logger.Error(q.ctx, "failed to close connIO", slog.Error(err)) |
| 1020 | } |
| 1021 | delete(q.mappers, mk) |
| 1022 | q.logger.Debug(q.ctx, "removed mapper", slog.F("peer_id", c.UniqueID())) |
| 1023 | } |
| 1024 | |
| 1025 | // maxBatchSize is the maximum number of keys to process in a single batch |
| 1026 | // query. |
no test coverage detected