Log retained wal bytes every 60 seconds

listener/listener.go

@@ -48,6 +48,7 @@ type replication interface {
 type repository interface {
 	CreatePublication(ctx context.Context, name string) error
 	GetSlotLSN(ctx context.Context, slotName string) (*string, error)
+	GetSlotRetainedWALBytes(ctx context.Context, slotName string) (*int64, error)
 	IsAlive() bool
 	Close(ctx context.Context) error
 }
@@ -382,6 +383,21 @@ func (l *Listener) Stream(ctx context.Context) error {
 		}
 	})
 
+	group.Go(func() error {
+		ticker := time.NewTicker(60 * time.Second)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ctx.Done():
+				l.log.Warn("stream: context canceled", "err", ctx.Err())
+				return nil
+			case <-ticker.C:
+				l.logRetainedWalBytes(ctx)
+			}
+		}
+	})
+
 	group.Go(func() error {
 		for {
 			if err := ctx.Err(); err != nil {
@@ -677,6 +693,21 @@ func (l *Listener) AckWalMessage(ctx context.Context, lsn pglogrepl.LSN) error {
 	return nil
 }
 
+func (l *Listener) logRetainedWalBytes(ctx context.Context) {
+	retainedWalBytes, err := l.repository.GetSlotRetainedWALBytes(ctx, l.cfg.Listener.SlotName)
+	if err != nil || retainedWalBytes == nil {
+		l.log.Error("failed to get retained WAL bytes", "err", err, slog.String("slot_name", l.cfg.Listener.SlotName))
+		return
+	}
+
+	l.log.Info(
+		"slot retained WAL bytes",
+		slog.String("lsn", l.readLSN().String()),
+		slog.String("slot_name", l.cfg.Listener.SlotName),
+		slog.Int64("retained_wal_bytes", *retainedWalBytes),
+	)
+}
+
 func (l *Listener) readLSN() pglogrepl.LSN {
 	l.mu.RLock()
 	defer l.mu.RUnlock()

listener/listener_test.go

@@ -3,6 +3,7 @@ package listener
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 	"strings"
@@ -618,3 +619,95 @@ func TestListener_Process(t *testing.T) {
 		})
 	}
 }
+
+func TestListener_logRetainedWalBytes(t *testing.T) {
+	repo := new(repositoryMock)
+
+	setGetSlotRetainedWALBytes := func(slotName string, bytes int64, err error) {
+		repo.On("GetSlotRetainedWALBytes", slotName).
+			Return(&bytes, err).
+			Once()
+	}
+
+	setGetSlotRetainedWALBytesNil := func(slotName string, err error) {
+		repo.On("GetSlotRetainedWALBytes", slotName).
+			Return((*int64)(nil), err).
+			Once()
+	}
+	tests := []struct {
+		name           string
+		setupMocks     func()
+		expectedLogMsg string
+		expectedLevel  string
+		expectedValue  string
+	}{
+		{
+			name: "successful log with retained WAL bytes",
+			setupMocks: func() {
+				setGetSlotRetainedWALBytes("test-slot", 1024000, nil)
+			},
+			expectedLogMsg: "slot retained WAL bytes",
+			expectedLevel:  "INFO",
+			expectedValue:  "1024000",
+		},
+		{
+			name: "error when query fails",
+			setupMocks: func() {
+				setGetSlotRetainedWALBytes("test-slot", 10, errors.New("database error"))
+			},
+			expectedLogMsg: "failed to get retained WAL bytes",
+			expectedLevel:  "ERROR",
+		},
+		{
+			name: "error when retained bytes is nil",
+			setupMocks: func() {
+				setGetSlotRetainedWALBytesNil("test-slot", nil)
+			},
+			expectedLogMsg: "failed to get retained WAL bytes",
+			expectedLevel:  "ERROR",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Setup
+			tt.setupMocks()
+
+			// Create a buffer to capture log output
+			var logBuffer strings.Builder
+			logger := slog.New(slog.NewJSONHandler(&logBuffer, &slog.HandlerOptions{
+				Level: slog.LevelDebug,
+			}))
+
+			// Create listener with test configuration
+			listener := &Listener{
+				log: logger,
+				cfg: &config.Config{
+					Listener: &config.ListenerCfg{
+						SlotName: "test-slot",
+					},
+				},
+				repository: repo,
+				lsn:        pglogrepl.LSN(12345), // Set a test LSN
+			}
+
+			// Execute
+			ctx := context.Background()
+			listener.logRetainedWalBytes(ctx)
+
+			// Verify
+			repo.AssertExpectations(t)
+
+			// Check log output
+			logOutput := logBuffer.String()
+			assert.Contains(t, logOutput, tt.expectedLogMsg)
+			assert.Contains(t, logOutput, tt.expectedLevel)
+
+			// For successful case, verify specific fields
+			if tt.expectedValue != "" {
+				assert.Contains(t, logOutput, `"slot_name":"test-slot"`)
+				assert.Contains(t, logOutput, fmt.Sprintf(`"retained_wal_bytes":%s`, tt.expectedValue))
+			}
+		})
+	}
+}

listener/repository.go

@@ -28,6 +28,16 @@ func (r RepositoryImpl) GetSlotLSN(ctx context.Context, slotName string) (*strin
 	return restartLSNStr, err
 }
 
+// GetSlotRetainedWALBytes returns the retained bytes of the replication slot.
+func (r RepositoryImpl) GetSlotRetainedWALBytes(ctx context.Context, slotName string) (*int64, error) {
+	var retainedWALBytes *int64
+
+	err := r.conn.QueryRow(ctx, "SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) AS retained_wal_bytes FROM pg_replication_slots WHERE slot_name=$1;", slotName).
+		Scan(&retainedWALBytes)
+
+	return retainedWALBytes, err
+}
+
 // CreatePublication create publication fo all.
 func (r RepositoryImpl) CreatePublication(ctx context.Context, name string) error {
 	if _, err := r.conn.Exec(ctx, `CREATE PUBLICATION "`+name+`" FOR ALL TABLES`); err != nil && !strings.Contains("already exists", err.Error()) {

listener/repository_mock_test.go

@@ -15,6 +15,11 @@ func (r *repositoryMock) GetSlotLSN(_ context.Context, slotName string) (*string
 	return args.Get(0).(*string), args.Error(1)
 }
 
+func (r *repositoryMock) GetSlotRetainedWALBytes(_ context.Context, slotName string) (*int64, error) {
+	args := r.Called(slotName)
+	return args.Get(0).(*int64), args.Error(1)
+}
+
 func (r *repositoryMock) IsAlive() bool {
 	return r.Called().Bool(0)
 }

nix/flake.nix

@@ -23,6 +23,7 @@
           packages = {
             direnv = pkgs.direnv;
             git = pkgs.git;
+            go = pkgs.go;
             google-cloud-sdk = pkgs.google-cloud-sdk.withExtraComponents ([
               pkgs.google-cloud-sdk.components.gke-gcloud-auth-plugin
               pkgs.google-cloud-sdk.components.pubsub-emulator