Merge pull request #7 from Trendyol/fix/not-updated-xlog-pos

fix: update xlog pos when no change in a while

connector.go

@@ -97,7 +97,9 @@ func NewConnector(ctx context.Context, cfg config.Config, listenerFunc replicati
 
 	m := metric.NewMetric(cfg.Slot.Name)
 
-	sl, err := slot.NewSlot(ctx, cfg.DSN(), cfg.Slot, m)
+	stream := replication.NewStream(conn, cfg, m, &system, listenerFunc)
+
+	sl, err := slot.NewSlot(ctx, cfg.DSN(), cfg.Slot, m, stream.(slot.XLogUpdater))
 	if err != nil {
 		return nil, err
 	}
@@ -108,7 +110,6 @@ func NewConnector(ctx context.Context, cfg config.Config, listenerFunc replicati
 	}
 	logger.Info("slot info", "info", slotInfo)
 
-	stream := replication.NewStream(conn, cfg, m, &system, listenerFunc)
 	prometheusRegistry := metric.NewRegistry(m)
 
 	tdb, err := timescaledb.NewTimescaleDB(ctx, cfg.DSN())

pq/replication/stream.go

@@ -5,6 +5,7 @@ import (
 	"encoding/binary"
 	goerrors "errors"
 	"fmt"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -55,6 +56,7 @@ type stream struct {
 	messageCH    chan *Message
 	listenerFunc ListenerFunc
 	sinkEnd      chan struct{}
+	mu           *sync.RWMutex
 	config       config.Config
 	lastXLogPos  pq.LSN
 	closed       atomic.Bool
@@ -71,6 +73,7 @@ func NewStream(conn pq.Connection, cfg config.Config, m metric.Metric, system *p
 		listenerFunc: listenerFunc,
 		lastXLogPos:  10,
 		sinkEnd:      make(chan struct{}, 1),
+		mu:           &sync.RWMutex{},
 	}
 }
 
@@ -123,7 +126,7 @@ func (s *stream) sink(ctx context.Context) {
 			}
 
 			if pgconn.Timeout(err) {
-				err = SendStandbyStatusUpdate(ctx, s.conn, uint64(0))
+				err = SendStandbyStatusUpdate(ctx, s.conn, uint64(s.LoadXLogPos()))
 				if err != nil {
 					logger.Error("send stand by status update", "error", err)
 					break
@@ -235,6 +238,21 @@ func (s *stream) GetMetric() metric.Metric {
 	return s.metric
 }
 
+func (s *stream) UpdateXLogPos(l pq.LSN) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if s.lastXLogPos < l {
+		s.lastXLogPos = l
+	}
+}
+
+func (s *stream) LoadXLogPos() pq.LSN {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.lastXLogPos
+}
+
 func SendStandbyStatusUpdate(_ context.Context, conn pq.Connection, walWritePosition uint64) error {
 	data := make([]byte, 0, 34)
 	data = append(data, StandbyStatusUpdateByteID)

pq/slot/slot.go

@@ -20,15 +20,20 @@ var (
 
 var typeMap = pgtype.NewMap()
 
+type XLogUpdater interface {
+	UpdateXLogPos(l pq.LSN)
+}
+
 type Slot struct {
-	conn      pq.Connection
-	metric    metric.Metric
-	ticker    *time.Ticker
-	statusSQL string
-	cfg       Config
+	conn       pq.Connection
+	metric     metric.Metric
+	logUpdater XLogUpdater
+	ticker     *time.Ticker
+	statusSQL  string
+	cfg        Config
 }
 
-func NewSlot(ctx context.Context, dsn string, cfg Config, m metric.Metric) (*Slot, error) {
+func NewSlot(ctx context.Context, dsn string, cfg Config, m metric.Metric, updater XLogUpdater) (*Slot, error) {
 	query := fmt.Sprintf("SELECT slot_name, slot_type, active, active_pid, restart_lsn, confirmed_flush_lsn, wal_status, PG_CURRENT_WAL_LSN() AS current_lsn FROM pg_replication_slots WHERE slot_name = '%s';", cfg.Name)
 
 	conn, err := pq.NewConnection(ctx, dsn)
@@ -37,11 +42,12 @@ func NewSlot(ctx context.Context, dsn string, cfg Config, m metric.Metric) (*Slo
 	}
 
 	return &Slot{
-		cfg:       cfg,
-		conn:      conn,
-		statusSQL: query,
-		metric:    m,
-		ticker:    time.NewTicker(time.Millisecond * cfg.SlotActivityCheckerInterval),
+		cfg:        cfg,
+		conn:       conn,
+		statusSQL:  query,
+		metric:     m,
+		ticker:     time.NewTicker(time.Millisecond * cfg.SlotActivityCheckerInterval),
+		logUpdater: updater,
 	}, nil
 }
 
@@ -109,6 +115,8 @@ func (s *Slot) Metrics(ctx context.Context) {
 		s.metric.SetSlotRetainedWALSize(float64(slotInfo.RetainedWALSize))
 		s.metric.SetSlotLag(float64(slotInfo.Lag))
 
+		s.logUpdater.UpdateXLogPos(slotInfo.CurrentLSN)
+
 		logger.Debug("slot metrics", "info", slotInfo)
 	}
 }