Compare commits

...

1 Commits

Author SHA1 Message Date
kasey
944f94a9bf recover from panics when writing the event stream (#14545)
* recover from panics when writing the event stream

---------

Co-authored-by: Kasey Kirkham <kasey@users.noreply.github.com>
(cherry picked from commit 1086bdf2b3)
2024-10-16 13:38:56 -05:00

View File

@@ -71,6 +71,7 @@ var (
errSlowReader = errors.New("client failed to read fast enough to keep outgoing buffer below threshold")
errNotRequested = errors.New("event not requested by client")
errUnhandledEventData = errors.New("unable to represent event data in the event stream")
errWriterUnusable = errors.New("http response writer is unusable")
)
// StreamingResponseWriter defines a type that can be used by the eventStreamer.
@@ -309,10 +310,21 @@ func (es *eventStreamer) outboxWriteLoop(ctx context.Context, cancel context.Can
}
}
func writeLazyReaderWithRecover(w StreamingResponseWriter, lr lazyReader) (err error) {
defer func() {
if r := recover(); r != nil {
log.WithField("panic", r).Error("Recovered from panic while writing event to client.")
err = errWriterUnusable
}
}()
_, err = io.Copy(w, lr())
return err
}
func (es *eventStreamer) writeOutbox(ctx context.Context, w StreamingResponseWriter, first lazyReader) error {
needKeepAlive := true
if first != nil {
if _, err := io.Copy(w, first()); err != nil {
if err := writeLazyReaderWithRecover(w, first); err != nil {
return err
}
needKeepAlive = false
@@ -325,13 +337,13 @@ func (es *eventStreamer) writeOutbox(ctx context.Context, w StreamingResponseWri
case <-ctx.Done():
return ctx.Err()
case rf := <-es.outbox:
if _, err := io.Copy(w, rf()); err != nil {
if err := writeLazyReaderWithRecover(w, rf); err != nil {
return err
}
needKeepAlive = false
default:
if needKeepAlive {
if _, err := io.Copy(w, newlineReader()); err != nil {
if err := writeLazyReaderWithRecover(w, newlineReader); err != nil {
return err
}
}