| 393 | } |
| 394 | |
| 395 | func (*Runner) handleStreamingResponse(ctx context.Context, logger slog.Logger, resp *http.Response) error { |
| 396 | buf := make([]byte, 4096) |
| 397 | totalRead := 0 |
| 398 | for { |
| 399 | // Check for context cancellation before each read |
| 400 | if ctx.Err() != nil { |
| 401 | logger.Warn(ctx, "streaming response canceled", |
| 402 | slog.F("bytes_read", totalRead), |
| 403 | slog.Error(ctx.Err()), |
| 404 | ) |
| 405 | return xerrors.Errorf("stream canceled: %w", ctx.Err()) |
| 406 | } |
| 407 | |
| 408 | n, err := resp.Body.Read(buf) |
| 409 | if n > 0 { |
| 410 | totalRead += n |
| 411 | } |
| 412 | if err == io.EOF { |
| 413 | break |
| 414 | } |
| 415 | if err != nil { |
| 416 | // Check if error is due to context cancellation |
| 417 | if xerrors.Is(err, context.Canceled) || xerrors.Is(err, context.DeadlineExceeded) { |
| 418 | logger.Warn(ctx, "streaming response read canceled", |
| 419 | slog.F("bytes_read", totalRead), |
| 420 | slog.Error(err), |
| 421 | ) |
| 422 | return xerrors.Errorf("stream read canceled: %w", err) |
| 423 | } |
| 424 | logger.Warn(ctx, "streaming response read error", |
| 425 | slog.F("bytes_read", totalRead), |
| 426 | slog.Error(err), |
| 427 | ) |
| 428 | return xerrors.Errorf("read stream: %w", err) |
| 429 | } |
| 430 | } |
| 431 | |
| 432 | logger.Debug(ctx, "received streaming response", slog.F("bytes_read", totalRead)) |
| 433 | return nil |
| 434 | } |
| 435 | |
| 436 | func (r *Runner) Cleanup(ctx context.Context, id string, logs io.Writer) error { |
| 437 | return r.strategy.Cleanup(ctx, id, logs) |