Merge branch 'feature/improve-sending-while-reconnecting' into featur…

…e/better-response-handler-errors

messagix/client.go

@@ -73,8 +73,8 @@ type Client struct {
 
 	stopCurrentConnection atomic.Pointer[context.CancelFunc]
 
-	CanSendMessages  bool
-	SendMessagesCond *sync.Cond
+	canSendMessages  bool
+	sendMessagesCond *sync.Cond
 }
 
 func NewClient(cookies *cookies.Cookies, logger zerolog.Logger) *Client {
@@ -98,8 +98,8 @@ func NewClient(cookies *cookies.Cookies, logger zerolog.Logger) *Client {
 		platform:         cookies.Platform,
 		activeTasks:      make([]int, 0),
 		taskMutex:        &sync.Mutex{},
-		CanSendMessages:  false,
-		SendMessagesCond: sync.NewCond(&sync.Mutex{}),
+		canSendMessages:  false,
+		sendMessagesCond: sync.NewCond(&sync.Mutex{}),
 	}
 	cli.http.CheckRedirect = cli.checkHTTPRedirect
 
@@ -225,6 +225,7 @@ func (c *Client) Connect() error {
 		connectionAttempts := 1
 		reconnectIn := 2 * time.Second
 		for {
+			c.disableSendingMessages() // In case we're reconnecting from a normal network error
 			connectStart := time.Now()
 			err := c.socket.Connect()
 			if ctx.Err() != nil {
@@ -393,3 +394,41 @@ func (c *Client) GetTaskId() int {
 	c.activeTasks = append(c.activeTasks, id)
 	return id
 }
+
+func (c *Client) EnableSendingMessages() {
+	c.sendMessagesCond.L.Lock()
+	c.canSendMessages = true
+	c.sendMessagesCond.Broadcast()
+	c.sendMessagesCond.L.Unlock()
+}
+
+func (c *Client) disableSendingMessages() {
+	c.sendMessagesCond.L.Lock()
+	c.canSendMessages = false
+	c.sendMessagesCond.L.Unlock()
+}
+
+func (c *Client) WaitUntilCanSendMessages(timeout time.Duration) error {
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+
+	done := make(chan struct{})
+	go func() {
+		c.sendMessagesCond.L.Lock()
+		defer c.sendMessagesCond.L.Unlock()
+		c.sendMessagesCond.Wait()
+		close(done)
+	}()
+
+	for !c.canSendMessages {
+		select {
+		case <-timer.C:
+			return fmt.Errorf("timeout waiting for canSendMessages")
+		case <-done:
+			if c.canSendMessages {
+				return nil
+			}
+		}
+	}
+	return nil
+}

messagix/events.go

@@ -110,10 +110,7 @@ func (s *Socket) handleReadyEvent(data *Event_Ready) error {
 	s.client.eventHandler(data.Finish())
 	s.previouslyConnected = true
 
-	s.client.SendMessagesCond.L.Lock()
-	s.client.CanSendMessages = true
-	s.client.SendMessagesCond.Signal()
-	s.client.SendMessagesCond.L.Unlock()
+	s.client.EnableSendingMessages()
 
 	return nil
 }

portal.go

@@ -54,6 +54,8 @@ import (
 	"go.mau.fi/mautrix-meta/msgconv"
 )
 
+const MaxMetaSendAttempts = 5
+
 func (br *MetaBridge) GetPortalByMXID(mxid id.RoomID) *Portal {
 	br.portalsLock.Lock()
 	defer br.portalsLock.Unlock()
@@ -630,12 +632,6 @@ func (portal *Portal) handleMatrixMessage(ctx context.Context, sender *User, evt
 	timings.convert = time.Since(start)
 	start = time.Now()
 
-	sender.Client.SendMessagesCond.L.Lock()
-	for !sender.Client.CanSendMessages {
-		sender.Client.SendMessagesCond.Wait()
-	}
-	sender.Client.SendMessagesCond.L.Unlock()
-
 	if waMsg != nil {
 		messageID := sender.E2EEClient.GenerateMessageID()
 		log.UpdateContext(func(c zerolog.Context) zerolog.Context {
@@ -657,10 +653,24 @@ func (portal *Portal) handleMatrixMessage(ctx context.Context, sender *User, evt
 		portal.pendingMessages[otid] = evt.ID
 		messageTS := time.Now()
 		var resp *table.LSTable
-		resp, err = sender.Client.ExecuteTasks(tasks...)
+
+		retries := 0
+		for retries < MaxMetaSendAttempts {
+			if err = sender.Client.WaitUntilCanSendMessages(15 * time.Second); err != nil {
+				log.Err(err).Msg("Error waiting to be able to send messages, retrying")
+			} else {
+				resp, err = sender.Client.ExecuteTasks(tasks...)
+				if err == nil {
+					break
+				}
+				log.Err(err).Msg("Failed to send message to Meta, retrying")
+			}
+			retries++
+		}
+
 		log.Trace().Any("response", resp).Msg("Meta send response")
 		var msgID string
-		if err == nil {
+		if resp != nil && err == nil {
 			for _, replace := range resp.LSReplaceOptimsiticMessage {
 				if replace.OfflineThreadingId == otidStr {
 					msgID = replace.MessageId