Skip to content
This repository has been archived by the owner on May 21, 2024. It is now read-only.

fix: omit reporting EOF errors #19

Merged
merged 3 commits into from
Jul 10, 2023
Merged

fix: omit reporting EOF errors #19

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
930fb5c
fix: omit reporting EOF errors
olegbespalov Jun 14, 2023
ff38c31
Apply suggestions from code review
olegbespalov Jun 14, 2023
5b3c8bc
chore: log only if there are no error handlers
olegbespalov Jun 14, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
4 changes: 3 additions & 1 deletion grpc/grpc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -117,12 +117,14 @@ func (mi *ModuleInstance) stream(c goja.ConstructorCall) *goja.Object {

p.SetSystemTags(mi.vu.State(), client.addr, methodName)

logger := mi.vu.State().Logger.WithField("streamMethod", methodName)

s := &stream{
vu: mi.vu,
client: client,
methodDescriptor: methodDescriptor,
method: methodName,
logger: mi.vu.State().Logger,
logger: logger,

tq: taskqueue.New(mi.vu.RegisterCallback),

Expand Down
51 changes: 33 additions & 18 deletions grpc/stream.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,12 +234,7 @@ func (s *stream) writeData(wg *sync.WaitGroup) {

err := s.stream.Send(msg.msg)
if err != nil {
s.logger.WithError(err).Error("failed to send data to the stream")

s.tq.Queue(func() error {
return s.closeWithError(err)
})

s.processSendError(err)
return
}

Expand Down Expand Up @@ -284,6 +279,17 @@ func (s *stream) writeData(wg *sync.WaitGroup) {
}
}

func (s *stream) processSendError(err error) {
if errors.Is(err, io.EOF) {
s.logger.WithError(err).Debug("skip sending a message stream is cancelled/finished")
err = nil
}
Comment on lines +283 to +286
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I would still prefer to just skip that ...

We can't know if this EOF is because that was what was expected.

If it isn't this is just shuffing it under the rug.
If it is - I feel the user should fix that. Maybe we can not emit it if the WebSocket (js object) is already closed, but doesn't seem like what the user reported to me.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If it isn't this is just shuffing it under the rug.

But according to to this, the real error will be returned with the RecvMsg. So EOF in sending context, is not the real error.

However, I now I'm thinking that we should not do closeWithError(nil) for that case 🤔 since it could hijack the real error reporting. 🤔

Copy link
Contributor

@codebien codebien Jun 15, 2023 •
edited

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

However, I now I'm thinking that we should not do closeWithError(nil)

In principle, it sounds correct to me

since it could hijack the real error reporting

but I'm not getting how this is possible. It sounds like closeWithError does not affect directly the read loop which is the place where it is expected to get the real error as you reported.

Or is it for the fact that closeWithError is called twice but the second is discarded?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Or is it for the fact that closeWithError is called twice but the second is discarded?

yeah, at the current state, closeWithError is designed to discard the second error if the stream is closed. So I think that I need to re-think this and maybe split the responsibilities to actually about the closing and error reporting.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So I think that I need to re-think this and maybe split the responsibilities to actually about the closing and error reporting.

What is the expectation, will it land in this PR?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hey @codebien, there are a couple of actions that I'm going to do, and they are on my list. No worries and no need to push this now.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay, I've pushed the commit where tried to address the issue that I've described above:

However, I now I'm thinking that we should not do closeWithError(nil) for that case thinking since it could hijack the real error reporting. thinking

Regarding the omitting the EOF (in sending), I still believe that's the right thing since as I linked to the grpc's code the real error if that happens is routing to the RecvMsg


s.tq.Queue(func() error {
return s.closeWithError(err)
})
}

// on registers a listener for a certain event type
func (s *stream) on(event string, listener func(goja.Value) (goja.Value, error)) {
if err := s.eventListeners.add(event, listener); err != nil {
Expand Down Expand Up @@ -323,10 +329,19 @@ func (s *stream) end() {
}

func (s *stream) closeWithError(err error) error {
s.close(err)

return s.callErrorListeners(err)
Comment on lines +332 to +334
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This means that a user might get multiple errors to be handled. Not certain if that is good or bad ...but I guess we can try it.

}

// close changes the stream state to closed and triggers the end event listeners
func (s *stream) close(err error) {
if s.state == closed {
return nil
return
}

s.logger.WithError(err).Debug("stream is closing")

s.state = closed
close(s.done)
s.tq.Queue(func() error {
Expand All @@ -336,24 +351,24 @@ func (s *stream) closeWithError(err error) error {
if s.timeoutCancel != nil {
s.timeoutCancel()
}

s.logger.WithError(err).Debug("connection closed")

if err != nil {
if errList := s.callErrorListeners(err); errList != nil {
return errList
}
}

return nil
}

func (s *stream) callErrorListeners(e error) error {
if e == nil {
return nil
}

rt := s.vu.Runtime()

obj := extractError(e)

for _, errorListener := range s.eventListeners.all(eventError) {
list := s.eventListeners.all(eventError)

if len(list) == 0 {
s.logger.Warnf("no handlers for error registered, but an error happened: %s", e)
}

for _, errorListener := range list {
if _, err := errorListener(rt.ToValue(obj)); err != nil {
return err
}
Expand Down