From 3440b731dfa0a4e64219641ca0c5e8e4daca71a8 Mon Sep 17 00:00:00 2001 From: Haw Loeung Date: Fri, 10 Jul 2026 19:11:50 +1000 Subject: [PATCH 1/6] Fix blocking channel send deadlock --- mm-go-irckit/user.go | 55 ++++++++++++++++++++++++++++++-------------- 1 file changed, 38 insertions(+), 17 deletions(-) diff --git a/mm-go-irckit/user.go b/mm-go-irckit/user.go index 0e972566..e7a1febd 100644 --- a/mm-go-irckit/user.go +++ b/mm-go-irckit/user.go @@ -167,63 +167,77 @@ func (u *User) Decode() { c := make(chan struct{}) <-c } - buffer := make(chan *irc.Message) + + buffer := make(chan *irc.Message, 512) stop := make(chan struct{}) + bufferTimeout := u.v.GetInt("PasteBufferTimeout") // we need at least 100 if bufferTimeout < 100 { bufferTimeout = 100 } logger.Debugf("using paste buffer timeout: %#v", bufferTimeout) + t := timer.NewTimer(time.Duration(bufferTimeout) * time.Millisecond) t.Stop() + go func(buffer chan *irc.Message, stop chan struct{}) { + var localBufferedMsg *irc.Message + for { select { - case msg := <-buffer: + case msg, ok := <-buffer: + if !ok { + return + } // are we starting a new buffer ? - if u.BufferedMsg == nil { - u.BufferedMsg = msg - // start timer now + if localBufferedMsg == nil { + localBufferedMsg = msg t.Reset(time.Duration(bufferTimeout) * time.Millisecond) } else { if strings.HasPrefix(msg.Trailing, "\x01ACTION") || replyRegExp.MatchString(msg.Trailing) || modifyRegExp.MatchString(msg.Trailing) { // flush buffer logger.Debug("flushing buffer because of /me, replies to threads, and message modifications") - u.BufferedMsg.Trailing = strings.TrimSpace(u.BufferedMsg.Trailing) - u.DecodeCh <- u.BufferedMsg - u.BufferedMsg = nil + localBufferedMsg.Trailing = strings.TrimSpace(localBufferedMsg.Trailing) + u.DecodeCh <- localBufferedMsg + localBufferedMsg = nil // send CTCP message u.DecodeCh <- msg continue } // make sure we're sending to the same recipient in the buffer - if u.BufferedMsg.Params[0] == msg.Params[0] { - u.BufferedMsg.Trailing += "\n" + msg.Trailing + if localBufferedMsg.Params[0] == msg.Params[0] { + localBufferedMsg.Trailing += "\n" + msg.Trailing } else { u.DecodeCh <- msg } } case <-t.C: - if u.BufferedMsg != nil { + if localBufferedMsg != nil { // trim last newline - u.BufferedMsg.Trailing = strings.TrimSpace(u.BufferedMsg.Trailing) - logger.Debugf("flushing buffer: %#v", u.BufferedMsg) - u.DecodeCh <- u.BufferedMsg + localBufferedMsg.Trailing = strings.TrimSpace(localBufferedMsg.Trailing) + logger.Debugf("flushing buffer: %#v", localBufferedMsg) + u.DecodeCh <- localBufferedMsg + localBufferedMsg = nil // clear buffer - u.BufferedMsg = nil t.Stop() } case <-stop: logger.Debug("closing decode()") + if localBufferedMsg != nil { + localBufferedMsg.Trailing = strings.TrimSpace(localBufferedMsg.Trailing) + u.DecodeCh <- localBufferedMsg + } return } } }(buffer, stop) + for { msg, err := u.Conn.Decode() if err != nil { close(stop) + close(buffer) // Signals the background worker that it's time to pack up isEOF := errors.Is(err, io.EOF) || err.Error() == "EOF" isClosed := errors.Is(err, net.ErrClosed) || strings.Contains(err.Error(), "use of closed network connection") @@ -239,16 +253,23 @@ func (u *User) Decode() { dmsg := "<- " + msg.String() if msg.Command == "PRIVMSG" && msg.Params != nil && (msg.Params[0] == "slack" || msg.Params[0] == "mattermost") { - // Don't log sensitive information trail := strings.Split(msg.Trailing, " ") if (msg.Trailing != "" && trail[0] == "login") || (len(msg.Params) > 1 && msg.Params[1] == "login") { dmsg = "<- PRIVMSG " + msg.Params[0] + " :login [redacted]" } } // 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") + u.DecodeCh <- msg + } } else { logger.Debug(dmsg) u.DecodeCh <- msg From d972ff371e493aafe82dd6ed96eb0d52e638a288 Mon Sep 17 00:00:00 2001 From: Haw Loeung Date: Fri, 10 Jul 2026 19:24:30 +1000 Subject: [PATCH 2/6] Ensure tickers are stopped --- bridge/mattermost/mattermost.go | 1 + mm-go-irckit/userbridge.go | 1 + 2 files changed, 2 insertions(+) diff --git a/bridge/mattermost/mattermost.go b/bridge/mattermost/mattermost.go index c4a25956..ebdb0408 100644 --- a/bridge/mattermost/mattermost.go +++ b/bridge/mattermost/mattermost.go @@ -140,6 +140,7 @@ func (m *Mattermost) loginToMattermost(onWsConnect func()) (*matterclient.Client //nolint:cyclop func (m *Mattermost) handleWsMessage(quitChan chan struct{}) { updateChannelsThrottle := time.NewTicker(time.Second * 60) + defer updateChannelsThrottle.Stop() for { if m.mc.WsQuit { diff --git a/mm-go-irckit/userbridge.go b/mm-go-irckit/userbridge.go index 18f1f3f3..5fbb2ea5 100644 --- a/mm-go-irckit/userbridge.go +++ b/mm-go-irckit/userbridge.go @@ -716,6 +716,7 @@ func (u *User) addUsersToChannels() { srv := u.Srv throttle := time.NewTicker(time.Millisecond * 200) + defer throttle.Stop() logger.Debug("in addUsersToChannels()") // add all users, also who are not on channels From 9d4041eda9457ccce5ff5f275b63ffa878602736 Mon Sep 17 00:00:00 2001 From: Haw Loeung Date: Fri, 10 Jul 2026 20:56:44 +1000 Subject: [PATCH 3/6] Improvements --- mm-go-irckit/user.go | 31 +++++++++++++++++++++++-------- 1 file changed, 23 insertions(+), 8 deletions(-) diff --git a/mm-go-irckit/user.go b/mm-go-irckit/user.go index e7a1febd..728625a9 100644 --- a/mm-go-irckit/user.go +++ b/mm-go-irckit/user.go @@ -181,6 +181,18 @@ func (u *User) Decode() { t := timer.NewTimer(time.Duration(bufferTimeout) * time.Millisecond) t.Stop() + safeSend := func(msg *irc.Message) bool { + t := time.NewTimer(2 * time.Second) + defer t.Stop() + select { + case u.DecodeCh <- msg: + return true + case <-t.C: + logger.Warnf("dropping message due to blocked DecodeCh: %#v", msg) + return false + } + } + go func(buffer chan *irc.Message, stop chan struct{}) { var localBufferedMsg *irc.Message @@ -199,17 +211,20 @@ func (u *User) Decode() { // flush buffer logger.Debug("flushing buffer because of /me, replies to threads, and message modifications") localBufferedMsg.Trailing = strings.TrimSpace(localBufferedMsg.Trailing) - u.DecodeCh <- localBufferedMsg + safeSend(localBufferedMsg) localBufferedMsg = nil // send CTCP message - u.DecodeCh <- msg + safeSend(msg) continue } // make sure we're sending to the same recipient in the buffer if localBufferedMsg.Params[0] == msg.Params[0] { localBufferedMsg.Trailing += "\n" + msg.Trailing } else { - u.DecodeCh <- msg + localBufferedMsg.Trailing = strings.TrimSpace(localBufferedMsg.Trailing) + safeSend(localBufferedMsg) + localBufferedMsg = msg + t.Reset(time.Duration(bufferTimeout) * time.Millisecond) } } case <-t.C: @@ -217,7 +232,7 @@ func (u *User) Decode() { // trim last newline localBufferedMsg.Trailing = strings.TrimSpace(localBufferedMsg.Trailing) logger.Debugf("flushing buffer: %#v", localBufferedMsg) - u.DecodeCh <- localBufferedMsg + safeSend(localBufferedMsg) localBufferedMsg = nil // clear buffer t.Stop() @@ -226,7 +241,7 @@ func (u *User) Decode() { logger.Debug("closing decode()") if localBufferedMsg != nil { localBufferedMsg.Trailing = strings.TrimSpace(localBufferedMsg.Trailing) - u.DecodeCh <- localBufferedMsg + safeSend(localBufferedMsg) } return } @@ -237,7 +252,6 @@ func (u *User) Decode() { msg, err := u.Conn.Decode() if err != nil { close(stop) - close(buffer) // Signals the background worker that it's time to pack up isEOF := errors.Is(err, io.EOF) || err.Error() == "EOF" isClosed := errors.Is(err, net.ErrClosed) || strings.Contains(err.Error(), "use of closed network connection") @@ -253,6 +267,7 @@ func (u *User) Decode() { dmsg := "<- " + msg.String() if msg.Command == "PRIVMSG" && msg.Params != nil && (msg.Params[0] == "slack" || msg.Params[0] == "mattermost") { + // Don't log sensitive information trail := strings.Split(msg.Trailing, " ") if (msg.Trailing != "" && trail[0] == "login") || (len(msg.Params) > 1 && msg.Params[1] == "login") { dmsg = "<- PRIVMSG " + msg.Params[0] + " :login [redacted]" @@ -268,11 +283,11 @@ func (u *User) Decode() { case buffer <- msg: default: logger.Warn("Paste buffer capacity breached, processing immediately to prevent thread hang") - u.DecodeCh <- msg + safeSend(msg) } } else { logger.Debug(dmsg) - u.DecodeCh <- msg + safeSend(msg) } } } From ec78f16d2153baa4a541e0797633f48def1fd901 Mon Sep 17 00:00:00 2001 From: Haw Loeung Date: Fri, 10 Jul 2026 21:51:55 +1000 Subject: [PATCH 4/6] Add safeEventSend --- bridge/mattermost/mattermost.go | 26 +++++++++++++++++++------- mm-go-irckit/user.go | 9 +++++---- 2 files changed, 24 insertions(+), 11 deletions(-) diff --git a/bridge/mattermost/mattermost.go b/bridge/mattermost/mattermost.go index ebdb0408..04fb9b3a 100644 --- a/bridge/mattermost/mattermost.go +++ b/bridge/mattermost/mattermost.go @@ -903,6 +903,18 @@ func (m *Mattermost) getParentReplyMsg(parentID string, newLen int, uncounted st return replyMessage, nil } +func (m *Mattermost) safeEventSend(event *bridge.Event) { + t := time.NewTimer(2 * time.Second) + defer t.Stop() + + select { + case m.eventChan <- event: + return + case <-t.C: + logger.Warnf("WARNING: m.eventChan is blocked! Dropping %s event to prevent worker deadlock: %s", event.Type, spew.Sdump(event)) + } +} + var ( validIRCNickRegExp = regexp.MustCompile("^[a-zA-Z0-9_]*$") channelMentionsRegExp = regexp.MustCompile(`@(channel|all|here)\W`) @@ -1015,7 +1027,7 @@ func (m *Mattermost) handleWsActionPost(rmsg *model.WebSocketEvent) { event.Data = d - m.eventChan <- event + m.safeEventSend(event) return } @@ -1028,7 +1040,7 @@ func (m *Mattermost) handleWsActionPost(rmsg *model.WebSocketEvent) { }, } - m.eventChan <- event + m.safeEventSend(event) return case model.PostTypeSystemGeneric, model.PostTypeEphemeral, model.PostTypeAddToTeam, model.PostTypeRemoveFromTeam: @@ -1130,7 +1142,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) { @@ -1151,7 +1163,7 @@ func (m *Mattermost) handleWsActionPost(rmsg *model.WebSocketEvent) { }, } - m.eventChan <- event + m.safeEventSend(event) } if len(data.FileIds) > 0 { @@ -1216,9 +1228,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) @@ -1481,7 +1493,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 728625a9..0c215098 100644 --- a/mm-go-irckit/user.go +++ b/mm-go-irckit/user.go @@ -181,15 +181,16 @@ func (u *User) Decode() { t := timer.NewTimer(time.Duration(bufferTimeout) * time.Millisecond) t.Stop() - safeSend := func(msg *irc.Message) bool { + safeSend := func(msg *irc.Message) { t := time.NewTimer(2 * time.Second) defer t.Stop() + select { case u.DecodeCh <- msg: - return true + return case <-t.C: - logger.Warnf("dropping message due to blocked DecodeCh: %#v", msg) - return false + logger.Warnf("WARNING: DecodeCh is blocked! Dropping message due to prevent worker deadlock: %#v", msg) + return } } From 80ea241b20f57841fb6f89e45039964cbae0743b Mon Sep 17 00:00:00 2001 From: Haw Loeung Date: Fri, 10 Jul 2026 22:35:22 +1000 Subject: [PATCH 5/6] Make it consistent with watchdog --- bridge/mattermost/mattermost.go | 25 ++++++++++++++++++------- mm-go-irckit/user.go | 20 +++++++++++--------- 2 files changed, 29 insertions(+), 16 deletions(-) diff --git a/bridge/mattermost/mattermost.go b/bridge/mattermost/mattermost.go index 04fb9b3a..edd86fc3 100644 --- a/bridge/mattermost/mattermost.go +++ b/bridge/mattermost/mattermost.go @@ -904,14 +904,25 @@ func (m *Mattermost) getParentReplyMsg(parentID string, newLen int, uncounted st } func (m *Mattermost) safeEventSend(event *bridge.Event) { - t := time.NewTimer(2 * time.Second) - defer t.Stop() + 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() - select { - case m.eventChan <- event: - return - case <-t.C: - logger.Warnf("WARNING: m.eventChan is blocked! Dropping %s event to prevent worker deadlock: %s", event.Type, spew.Sdump(event)) + m.eventChan <- event + + if elapsed := time.Since(start); elapsed > 5*time.Second { + logger.Warnf( + "eventChan send recovered after %v (type=%s)", + elapsed, + event.Type, + ) } } diff --git a/mm-go-irckit/user.go b/mm-go-irckit/user.go index 0c215098..4c3f0884 100644 --- a/mm-go-irckit/user.go +++ b/mm-go-irckit/user.go @@ -182,15 +182,17 @@ func (u *User) Decode() { t.Stop() safeSend := func(msg *irc.Message) { - t := time.NewTimer(2 * time.Second) - defer t.Stop() - - select { - case u.DecodeCh <- msg: - return - case <-t.C: - logger.Warnf("WARNING: DecodeCh is blocked! Dropping message due to prevent worker deadlock: %#v", msg) - return + 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) } } From 3f8543f6f07f257d22ca6a24e1c1337b3c284ab8 Mon Sep 17 00:00:00 2001 From: Haw Loeung Date: Sat, 11 Jul 2026 08:11:14 +1000 Subject: [PATCH 6/6] Ensure ticker/timer is cleaned up --- mm-go-irckit/userbridge.go | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/mm-go-irckit/userbridge.go b/mm-go-irckit/userbridge.go index 5fbb2ea5..6de09965 100644 --- a/mm-go-irckit/userbridge.go +++ b/mm-go-irckit/userbridge.go @@ -716,7 +716,6 @@ func (u *User) addUsersToChannels() { srv := u.Srv throttle := time.NewTicker(time.Millisecond * 200) - defer throttle.Stop() logger.Debug("in addUsersToChannels()") // add all users, also who are not on channels @@ -742,8 +741,14 @@ func (u *User) addUsersToChannels() { ch.Join(u) channels := make(chan *bridge.ChannelInfo, 5) + var wg sync.WaitGroup + for i := 0; i < 10; i++ { - go u.addUserToChannelWorker(channels, throttle) + wg.Add(1) + go func() { + defer wg.Done() + u.addUserToChannelWorker(channels, throttle) + }() } for _, brchannel := range u.br.GetChannels() { @@ -761,6 +766,12 @@ func (u *User) addUsersToChannels() { close(channels) + // clean up tickers/timers after workers have finished + go func() { + wg.Wait() + throttle.Stop() + }() + // we did all the initialization, now listen for events go u.handleEventChan() }