Add a user-facing log when processing large txn (#4500)

Add a user-facing log when a large txn result in a PullRecord sync
interval to get extended.

Logged once per sync_interval, but throttled at 1 minute in case
customer has a very short sync intervals that makes the log excessively
noisy.

Fixes: DBI-681

Author: jgao54

flow/activities/flowable.go

@@ -583,7 +583,7 @@ func (a *FlowableActivity) GetQRepPartitions(ctx context.Context,
 			if bytes > 100<<30 { // 100 GiB
 				msg := fmt.Sprintf("large table detected: %s (%s). Counting/partitioning queries for parallel "+
 					"snapshotting may take minutes to hours to execute. This is normal for tables over 100 GiB.",
-					config.WatermarkTable, utils.FormatTableSize(bytes))
+					config.WatermarkTable, utils.HumanReadableBytes(bytes))
 				a.Alerter.LogFlowInfo(ctx, config.FlowJobName, msg)
 			}
 		} else {

flow/connectors/bigquery/source_schema.go

@@ -130,7 +130,7 @@ func (c *BigQueryConnector) GetTablesInSchema(ctx context.Context, schema string
 				slog.Any("error", err))
 			tableSize = "Unknown"
 		} else {
-			tableSize = utils.FormatTableSize(metadata.NumBytes)
+			tableSize = utils.HumanReadableBytes(metadata.NumBytes)
 		}
 
 		tableResponse := &protos.TableResponse{

flow/connectors/postgres/cdc.go

@@ -25,6 +25,7 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	"go.temporal.io/sdk/log"
 
+	"github.com/PeerDB-io/peerdb/flow/alerting"
 	connmetadata "github.com/PeerDB-io/peerdb/flow/connectors/external_metadata"
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils"
 	"github.com/PeerDB-io/peerdb/flow/connectors/utils/monitoring"
@@ -59,6 +60,7 @@ type PostgresCDCSource struct {
 
 	// for storing schema delta audit logs to catalog
 	catalogPool                              shared.CatalogPool
+	alerter                                  *alerting.Alerter
 	otelManager                              *otel_metrics.OtelManager
 	hushWarnUnhandledMessageType             map[pglogrepl.MessageType]struct{}
 	hushWarnUnknownTableDetected             map[uint32]struct{}
@@ -130,6 +132,7 @@ func (c *PostgresConnector) NewPostgresCDCSource(ctx context.Context, cdcConfig
 		idToRelKindMap:                           idToRelKindMap,
 		publishViaPartitionRoot:                  publishViaPartitionRoot,
 		catalogPool:                              cdcConfig.CatalogPool,
+		alerter:                                  alerting.NewAlerter(ctx, cdcConfig.CatalogPool, cdcConfig.OtelManager),
 		otelManager:                              cdcConfig.OtelManager,
 		hushWarnUnhandledMessageType:             make(map[pglogrepl.MessageType]struct{}),
 		hushWarnUnknownTableDetected:             make(map[uint32]struct{}),
@@ -584,6 +587,7 @@ func PullCdcRecords[Items model.Items](
 	defer shutdown()
 
 	var standByLastLogged time.Time
+	var largeTxnLastLogged time.Time
 	nextRecordDeadline := time.Now().Add(req.IdleTimeout)
 	pkmRequiresResponse := false
 
@@ -679,11 +683,24 @@ func PullCdcRecords[Items model.Items](
 						slog.Float64("elapsedMinutes", time.Since(pullStart).Minutes()))
 					return nil
 				} else {
+					accumulatedBytes := totalFetchedBytes.Load()
 					logger.Info("commit lock, waiting for commit to return records",
 						slog.Int64("records", totalRecords),
-						slog.Int64("bytes", totalFetchedBytes.Load()),
+						slog.Int64("bytes", accumulatedBytes),
 						slog.Int("channelLen", records.ChannelLen()),
 						slog.Float64("elapsedMinutes", time.Since(pullStart).Minutes()))
+
+					if time.Since(largeTxnLastLogged) > time.Minute {
+						if !largeTxnLastLogged.IsZero() {
+							userMsg := fmt.Sprintf(
+								"Reading a large already-committed transaction off the replication slot; "+
+									"sync will continue once it's fully received (%d records, %s buffered so far).",
+								totalRecords, utils.HumanReadableBytes(accumulatedBytes))
+							p.alerter.LogFlowInfo(ctx, p.flowJobName, userMsg)
+						}
+						largeTxnLastLogged = time.Now()
+					}
+
 					waitingForCommit = true
 				}
 			} else {

flow/connectors/utils/formatters.go

@@ -4,8 +4,8 @@ import "fmt"
 
 var units = []string{"B", "KB", "MB", "GB", "TB", "PB"}
 
-// FormatTableSize converts bytes to human-readable format
-func FormatTableSize(bytes int64) string {
+// HumanReadableBytes converts bytes to human-readable format
+func HumanReadableBytes(bytes int64) string {
 	if bytes == 0 {
 		return "0 B"
 	}