Wrap events and IRC messages to clients watchdog and log

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 {

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)
 		}
 	}
 }