| 963 | } |
| 964 | |
| 965 | func (q *querier) newConn(c *connIO) { |
| 966 | q.mu.Lock() |
| 967 | defer q.mu.Unlock() |
| 968 | if !q.healthy { |
| 969 | _ = c.Enqueue(&proto.CoordinateResponse{Error: CloseErrUnhealthy}) |
| 970 | err := c.Close() |
| 971 | // This can only happen during a narrow window where we were healthy |
| 972 | // when pgCoord checked before accepting the connection, but now are |
| 973 | // unhealthy now that we get around to processing it. Seeing a small |
| 974 | // number of these logs is not worrying, but a large number probably |
| 975 | // indicates something is amiss. |
| 976 | q.logger.Warn(q.ctx, "closed incoming connection while unhealthy", |
| 977 | slog.Error(err), |
| 978 | slog.F("peer_id", c.UniqueID()), |
| 979 | ) |
| 980 | return |
| 981 | } |
| 982 | mpr := newMapper(c, q.logger, q.heartbeats) |
| 983 | mk := mKey(c.UniqueID()) |
| 984 | dup, ok := q.mappers[mk] |
| 985 | if ok { |
| 986 | q.logger.Debug(q.ctx, "duplicate mapper found; closing old connection", slog.F("peer_id", dup.c.UniqueID())) |
| 987 | err := dup.c.CoordinatorClose() |
| 988 | if err != nil { |
| 989 | q.logger.Error(q.ctx, "failed to close duplicate mapper", slog.F("peer_id", dup.c.UniqueID()), slog.Error(err)) |
| 990 | } |
| 991 | } |
| 992 | q.mappers[mk] = mpr |
| 993 | q.mappingQ.enqueue(mk) |
| 994 | q.logger.Debug(q.ctx, "added new mapper", slog.F("peer_id", c.UniqueID())) |
| 995 | } |
| 996 | |
| 997 | func (q *querier) isHealthy() bool { |
| 998 | q.mu.Lock() |