Close the coordination gracefully. If the context expires before the remote API server has hung up on us, we forcibly close the Client connection.
(ctx context.Context)
| 249 | // Close the coordination gracefully. If the context expires before the remote API server has hung |
| 250 | // up on us, we forcibly close the Client connection. |
| 251 | func (c *BasicCoordination) Close(ctx context.Context) (retErr error) { |
| 252 | c.Lock() |
| 253 | if c.closed { |
| 254 | c.Unlock() |
| 255 | return nil |
| 256 | } |
| 257 | c.closed = true |
| 258 | err := c.client.Send(&proto.CoordinateRequest{Disconnect: &proto.CoordinateRequest_Disconnect{}}) |
| 259 | c.Unlock() |
| 260 | if err != nil && !xerrors.Is(err, io.EOF) { |
| 261 | reason := codersdk.DisconnectReasonNetworkError |
| 262 | // Log but don't return early; we must still clean up below. |
| 263 | c.logger.Warn(context.Background(), "failed to send disconnect", |
| 264 | c.direction.SlogField(), |
| 265 | reason.SlogExpectedField(), |
| 266 | reason.SlogField(), |
| 267 | c.initiator.SlogField(), |
| 268 | slog.Error(err), |
| 269 | ) |
| 270 | retErr = xerrors.Errorf("send disconnect: %w", err) |
| 271 | } else { |
| 272 | reason := codersdk.DisconnectReasonGraceful |
| 273 | c.logger.Debug(context.Background(), "sent disconnect", |
| 274 | c.direction.SlogField(), |
| 275 | reason.SlogExpectedField(), |
| 276 | reason.SlogField(), |
| 277 | c.initiator.SlogField(), |
| 278 | ) |
| 279 | } |
| 280 | |
| 281 | // We shouldn't just close the protocol right away, because the way dRPC streams work is |
| 282 | // that if you close them, that could take effect immediately, even before the Disconnect |
| 283 | // message is processed. Coordinators are supposed to hang up on us once they get a |
| 284 | // Disconnect message, so we should wait around for that until the context expires. |
| 285 | select { |
| 286 | case <-c.respLoopDone: |
| 287 | reason := codersdk.DisconnectReasonGraceful |
| 288 | c.logger.Debug(ctx, "responses closed after disconnect", |
| 289 | c.direction.SlogField(), |
| 290 | reason.SlogExpectedField(), |
| 291 | reason.SlogField(), |
| 292 | c.initiator.SlogField(), |
| 293 | ) |
| 294 | return retErr |
| 295 | case <-ctx.Done(): |
| 296 | reason := codersdk.DisconnectReasonNetworkError |
| 297 | c.logger.Warn(ctx, "context expired while waiting for coordinate responses to close", |
| 298 | c.direction.SlogField(), |
| 299 | reason.SlogExpectedField(), |
| 300 | reason.SlogField(), |
| 301 | c.initiator.SlogField(), |
| 302 | codersdk.SlogDisconnectDetail("context expired before coordinator hung up"), |
| 303 | ) |
| 304 | } |
| 305 | // forcefully close the stream |
| 306 | protoErr := c.client.Close() |
| 307 | <-c.respLoopDone |
| 308 | if retErr == nil { |
nothing calls this directly
no test coverage detected