diff --git a/bridge/mattermost/mattermost.go b/bridge/mattermost/mattermost.go index ed47503c..68d843ea 100644 --- a/bridge/mattermost/mattermost.go +++ b/bridge/mattermost/mattermost.go @@ -903,6 +903,29 @@ func (m *Mattermost) getParentReplyMsg(parentID string, newLen int, uncounted st return replyMessage, nil } +func (m *Mattermost) safeEventSend(event *bridge.Event) { + start := time.Now() + + timer := time.AfterFunc(5*time.Second, func() { + logger.Warnf( + "eventChan send blocked for >5s (type=%s): %s", + event.Type, + spew.Sdump(event), + ) + }) + defer timer.Stop() + + m.eventChan <- event + + if elapsed := time.Since(start); elapsed > 5*time.Second { + logger.Warnf( + "eventChan send recovered after %v (type=%s)", + elapsed, + event.Type, + ) + } +} + var ( validIRCNickRegExp = regexp.MustCompile("^[a-zA-Z0-9_]*$") channelMentionsRegExp = regexp.MustCompile(`@(channel|all|here)\W`) @@ -1000,7 +1023,7 @@ func (m *Mattermost) handleWsActionPost(rmsg *model.WebSocketEvent) { Type: "direct_message", Data: d, } - m.eventChan <- event + m.safeEventSend(event) } switch data.Type { @@ -1031,7 +1054,7 @@ func (m *Mattermost) handleWsActionPost(rmsg *model.WebSocketEvent) { }, } - m.eventChan <- event + m.safeEventSend(event) return default: @@ -1060,7 +1083,7 @@ func (m *Mattermost) handleWsActionPost(rmsg *model.WebSocketEvent) { }, } - m.eventChan <- event + m.safeEventSend(event) return } @@ -1157,7 +1180,7 @@ func (m *Mattermost) handleWsActionPost(rmsg *model.WebSocketEvent) { event.Data = d - m.eventChan <- event + m.safeEventSend(event) default: messageType := "" if !m.v.GetBool("mattermost.disabledefaultmentions") && channelMentionsRegExp.MatchString(data.Message) { @@ -1178,7 +1201,7 @@ func (m *Mattermost) handleWsActionPost(rmsg *model.WebSocketEvent) { }, } - m.eventChan <- event + m.safeEventSend(event) } if len(data.FileIds) > 0 { @@ -1243,9 +1266,9 @@ func (m *Mattermost) handleFileEvent(channelType string, ghost *bridge.UserInfo, return } - m.eventChan <- event + m.safeEventSend(event) default: - m.eventChan <- event + m.safeEventSend(event) } logger.Debugf("handleFileEvent() user %s sent %d files %#v", ghost.Nick, len(fileEvent.Files), data.FileIds) @@ -1508,7 +1531,7 @@ func (m *Mattermost) handleReactionEvent(rmsg *model.WebSocketEvent) { } } - m.eventChan <- event + m.safeEventSend(event) } func (m *Mattermost) GetTeamName(teamID string) string { diff --git a/mm-go-irckit/user.go b/mm-go-irckit/user.go index 0f164fd6..531b92b8 100644 --- a/mm-go-irckit/user.go +++ b/mm-go-irckit/user.go @@ -179,6 +179,22 @@ func (u *User) Decode() { bufferTimeout = 100 } logger.Debugf("using paste buffer timeout: %#v", bufferTimeout) + + safeSend := func(msg *irc.Message) { + start := time.Now() + + timer := time.AfterFunc(5*time.Second, func() { + logger.Warnf("DecodeCh send blocked for >5s: %#v", msg) + }) + defer timer.Stop() + + u.DecodeCh <- msg + + if elapsed := time.Since(start); elapsed > 5*time.Second { + logger.Warnf("DecodeCh send recovered after %v", elapsed) + } + } + timeout := time.Duration(bufferTimeout) * time.Millisecond t := timer.NewTimer(timeout) t.Stop() @@ -232,7 +248,7 @@ func (u *User) Decode() { logger.Debug("flushing buffer because of /me, replies to threads, and message modifications") flush() // send CTCP message - u.DecodeCh <- msg + safeSend(msg) continue } // make sure we're sending to the same recipient in the buffer @@ -276,12 +292,20 @@ func (u *User) Decode() { } } // PRIVMSG can be buffered + if msg.Command == "PRIVMSG" { logger.Debugf("B: %#v", dmsg) - buffer <- msg + + // Safe non-blocking channel push guard + select { + case buffer <- msg: + default: + logger.Warn("Paste buffer capacity breached, processing immediately to prevent thread hang") + safeSend(msg) + } } else { logger.Debug(dmsg) - u.DecodeCh <- msg + safeSend(msg) } } }