From 6006ca8fe5178ce231d14f52f1b6b10438fbbae1 Mon Sep 17 00:00:00 2001
From: Morgan Tocker <mtocker@squareup.com>
Date: Fri, 27 Sep 2024 08:29:29 -0600
Subject: [PATCH] repl: Improve Flush() to handle BlockWait() timeouts

---
 pkg/repl/client.go | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/pkg/repl/client.go b/pkg/repl/client.go
index 4be9be9..d022f9b 100644
--- a/pkg/repl/client.go
+++ b/pkg/repl/client.go
@@ -641,9 +641,14 @@ func (c *Client) Flush(ctx context.Context) error {
 			return err
 		}
 		// Wait for canal to catch up before determining if the changeset
-		// length is considered trivial.
+		// length is considered trivial. If it can't catch up before the
+		// timeout is reached (default 10s), it will return an error.
+		// In which case we are fine to just continue, which will start
+		// the loop again. After we flush what new changes we discovered here,
+		// we can try BlockWaiting again.
 		if err := c.BlockWait(ctx); err != nil {
-			return err
+			c.logger.Warnf("error waiting for canal to catch up: %v", err)
+			continue
 		}
 		if c.GetDeltaLen() < binlogTrivialThreshold {
 			break