last

source/config.go

@@ -53,7 +53,7 @@ type Config struct {
 	URL string `json:"url" validate:"required"`
 	// Tables is a List of table names to read from, separated by a comma, e.g.:"table1,table2".
 	// Use "*" if you'd like to listen to all tables.
-	Tables []string `json:"tables" validate:"required"`
+	Tables []string `json:"tables"`
 	// Deprecated: use `tables` instead.
 	Table []string `json:"table"`
 

source/logrepl/cdc.go

@@ -21,6 +21,7 @@ import (
 	"sync/atomic"
 
 	"github.com/conduitio/conduit-connector-postgres/source/logrepl/internal"
+	"github.com/conduitio/conduit-connector-postgres/source/position"
 	sdk "github.com/conduitio/conduit-connector-sdk"
 	"github.com/jackc/pglogrepl"
 	"github.com/jackc/pgx/v5/pgconn"
@@ -35,19 +36,6 @@ type CDCConfig struct {
 	TableKeys       map[string]string
 }
 
-// Validate performs validation tasks on the config.
-func (c Config) Validate() error {
-	var errs []error
-	// make sure we have all table keys
-	for _, tableName := range c.Tables {
-		if c.TableKeys[tableName] == "" {
-			errs = append(errs, fmt.Errorf("missing key for table %q", tableName))
-		}
-	}
-
-	return errors.Join(errs...)
-}
-
 // CDCIterator asynchronously listens for events from the logical replication
 // slot and returns them to the caller through Next.
 type CDCIterator struct {
@@ -164,11 +152,25 @@ func (i *CDCIterator) Next(ctx context.Context) (sdk.Record, error) {
 }
 
 // Ack forwards the acknowledgment to the subscription.
-func (i *CDCIterator) Ack(_ context.Context, pos sdk.Position) error {
-	lsn, err := PositionToLSN(pos)
+func (i *CDCIterator) Ack(_ context.Context, sdkPos sdk.Position) error {
+	pos, err := position.ParseSDKPosition(sdkPos)
+	if err != nil {
+		return err
+	}
+
+	if pos.Type != position.TypeCDC {
+		return fmt.Errorf("invalid type %q for CDC position", pos.Type.String())
+	}
+
+	lsn, err := pos.LSN()
 	if err != nil {
-		return fmt.Errorf("failed to parse position: %w", err)
+		return err
+	}
+
+	if lsn == 0 {
+		return fmt.Errorf("cannot ack zero position")
 	}
+
 	i.sub.Ack(lsn)
 	return nil
 }
@@ -177,20 +179,21 @@ func (i *CDCIterator) Ack(_ context.Context, pos sdk.Position) error {
 // or the context gets canceled. If the subscription stopped with an unexpected
 // error, the error is returned.
 func (i *CDCIterator) Teardown(ctx context.Context) error {
-	// Close after Subscription is done,
-	// connection may still be in use.
 	defer i.pgconn.Close(ctx)
 
+	if started := i.subStarted.Load(); !started {
+		return nil
+	}
+
 	i.sub.Stop()
 
 	select {
 	case <-ctx.Done():
+		// When context is done, the pgconn will be closed hastedly
+		// without waiting for the subscription to be completely stopped.
 		return ctx.Err()
 	case <-i.sub.Ready():
 		return i.sub.Wait(ctx)
-	default:
-		// subscription was never started, nothing to do
-		return nil
 	}
 }
 
@@ -201,6 +204,8 @@ func (i *CDCIterator) TXSnapshotID() string {
 	return i.sub.TXSnapshotID
 }
 
+// withReplication adds the `replication` parameter to the connection config.
+// This will uprgade a regular command connection to accept replication commands.
 func withReplication(pgconf *pgconn.Config) *pgconn.Config {
 	c := pgconf.Copy()
 	if c.RuntimeParams == nil {

source/logrepl/cdc_test.go

@@ -16,23 +16,125 @@ package logrepl
 
 import (
 	"context"
+	"errors"
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/conduitio/conduit-connector-postgres/source/position"
 	"github.com/conduitio/conduit-connector-postgres/test"
 	sdk "github.com/conduitio/conduit-connector-sdk"
+	"github.com/jackc/pgx/v5/pgconn"
 	"github.com/jackc/pgx/v5/pgxpool"
 	"github.com/matryer/is"
 )
 
-func TestIterator_Next(t *testing.T) {
+func TestCDCIterator_New(t *testing.T) {
+	ctx := context.Background()
+	pool := test.ConnectPool(ctx, t, test.RepmgrConnString)
+
+	tests := []struct {
+		name    string
+		setup   func(t *testing.T) CDCConfig
+		pgconf  *pgconn.Config
+		wantErr error
+	}{
+		{
+			name:   "publication already exists",
+			pgconf: &pool.Config().ConnConfig.Config,
+			setup: func(t *testing.T) CDCConfig {
+				is := is.New(t)
+				table := test.SetupTestTable(ctx, t, pool)
+				test.CreatePublication(t, pool, table, []string{table})
+
+				t.Cleanup(func() {
+					is.NoErr(Cleanup(ctx, CleanupConfig{
+						URL:      pool.Config().ConnString(),
+						SlotName: table,
+					}))
+				})
+
+				return CDCConfig{
+					SlotName:        table,
+					PublicationName: table,
+					Tables:          []string{table},
+				}
+			},
+		},
+		{
+			name: "fails to connect",
+			pgconf: func() *pgconn.Config {
+				c := pool.Config().ConnConfig.Config
+				c.Port = 31337
+
+				return &c
+			}(),
+			setup: func(*testing.T) CDCConfig {
+				return CDCConfig{}
+			},
+			wantErr: errors.New("could not establish replication connection"),
+		},
+		{
+			name:   "fails to create publication",
+			pgconf: &pool.Config().ConnConfig.Config,
+			setup: func(*testing.T) CDCConfig {
+				return CDCConfig{
+					PublicationName: "foobar",
+				}
+			},
+			wantErr: errors.New("requires at least one table"),
+		},
+		{
+			name:   "fails to create subscription",
+			pgconf: &pool.Config().ConnConfig.Config,
+			setup: func(t *testing.T) CDCConfig {
+				is := is.New(t)
+				table := test.SetupTestTable(ctx, t, pool)
+
+				t.Cleanup(func() {
+					is.NoErr(Cleanup(ctx, CleanupConfig{
+						URL:             pool.Config().ConnString(),
+						PublicationName: table,
+					}))
+				})
+
+				return CDCConfig{
+					SlotName:        "invalid,name_/",
+					PublicationName: table,
+					Tables:          []string{table},
+				}
+			},
+			wantErr: errors.New("ERROR: syntax error (SQLSTATE 42601)"),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			is := is.New(t)
+
+			config := tt.setup(t)
+
+			_, err := NewCDCIterator(ctx, tt.pgconf, config)
+			if tt.wantErr != nil {
+				if match := strings.Contains(err.Error(), tt.wantErr.Error()); !match {
+					t.Logf("%s != %s", err.Error(), tt.wantErr.Error())
+					is.True(match)
+				}
+			} else {
+				is.NoErr(err)
+			}
+		})
+	}
+}
+
+func TestCDCIterator_Next(t *testing.T) {
 	ctx := context.Background()
 	is := is.New(t)
 
 	pool := test.ConnectPool(ctx, t, test.RepmgrConnString)
 	table := test.SetupTestTable(ctx, t, pool)
-	i := testIterator(ctx, t, pool, table)
+	i := testCDCIterator(ctx, t, pool, table, true)
 
 	// wait for subscription to be ready
 	<-i.sub.Ready()
@@ -132,7 +234,122 @@ func TestIterator_Next(t *testing.T) {
 	}
 }
 
-func testIterator(ctx context.Context, t *testing.T, pool *pgxpool.Pool, table string) *CDCIterator {
+func TestCDCIterator_Next_Fail(t *testing.T) {
+	ctx := context.Background()
+
+	pool := test.ConnectPool(ctx, t, test.RepmgrConnString)
+	table := test.SetupTestTable(ctx, t, pool)
+
+	t.Run("fail when sub is done", func(t *testing.T) {
+		is := is.New(t)
+
+		i := testCDCIterator(ctx, t, pool, table, true)
+		<-i.sub.Ready()
+
+		is.NoErr(i.Teardown(ctx))
+
+		_, err := i.Next(ctx)
+		expectErr := "logical replication error:"
+
+		match := strings.Contains(err.Error(), expectErr)
+		if !match {
+			t.Logf("%s != %s", err.Error(), expectErr)
+		}
+		is.True(match)
+	})
+
+	t.Run("fail when subscriber is not started", func(t *testing.T) {
+		is := is.New(t)
+
+		i := testCDCIterator(ctx, t, pool, table, false)
+
+		_, nexterr := i.Next(ctx)
+		is.Equal(nexterr.Error(), "logical replication has not been started")
+	})
+}
+
+func TestCDCIterator_Ack(t *testing.T) {
+	ctx := context.Background()
+
+	tests := []struct {
+		name    string
+		pos     sdk.Position
+		wantErr error
+	}{
+		{
+			name:    "failed to parse position",
+			pos:     sdk.Position([]byte("{")),
+			wantErr: errors.New("invalid position: unexpected end of JSON input"),
+		},
+		{
+			name: "position of wrong type",
+			pos: position.Position{
+				Type: position.TypeSnapshot,
+			}.ToSDKPosition(),
+			wantErr: errors.New(`invalid type "Snapshot" for CDC position`),
+		},
+		{
+			name: "failed to parse LSN",
+			pos: position.Position{
+				Type:    position.TypeCDC,
+				LastLSN: "garble",
+			}.ToSDKPosition(),
+			wantErr: errors.New("failed to parse LSN: expected integer"),
+		},
+		{
+			name: "invalid position LSN",
+			pos: position.Position{
+				Type: position.TypeCDC,
+			}.ToSDKPosition(),
+			wantErr: errors.New("cannot ack zero position"),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			is := is.New(t)
+
+			i := &CDCIterator{}
+
+			err := i.Ack(ctx, tt.pos)
+			if tt.wantErr != nil {
+				is.Equal(err.Error(), tt.wantErr.Error())
+			} else {
+				is.NoErr(err)
+			}
+		})
+	}
+}
+
+func TestCDCIterator_Teardown(t *testing.T) {
+	t.Skip("This causes a data race in the the postgres connection being closed during forced teardown")
+
+	ctx := context.Background()
+	is := is.New(t)
+
+	pool := test.ConnectPool(ctx, t, test.RepmgrConnString)
+	table := test.SetupTestTable(ctx, t, pool)
+	i := testCDCIterator(ctx, t, pool, table, true)
+
+	// wait for subscription to be ready
+	<-i.sub.Ready()
+
+	cctx, cancel := context.WithCancel(ctx)
+	cancel()
+
+	_ = i.Teardown(cctx)
+	// is.Equal(err.Error(), "context canceled")
+	_ = is
+}
+
+func Test_withReplication(t *testing.T) {
+	is := is.New(t)
+
+	c := withReplication(&pgconn.Config{})
+	is.Equal(c.RuntimeParams["replication"], "database")
+}
+
+func testCDCIterator(ctx context.Context, t *testing.T, pool *pgxpool.Pool, table string, start bool) *CDCIterator {
 	is := is.New(t)
 	config := CDCConfig{
 		Tables:          []string{table},
@@ -143,6 +360,11 @@ func testIterator(ctx context.Context, t *testing.T, pool *pgxpool.Pool, table s
 
 	i, err := NewCDCIterator(ctx, &pool.Config().ConnConfig.Config, config)
 	is.NoErr(err)
+
+	if start {
+		is.NoErr(i.StartSubscriber(ctx))
+	}
+
 	t.Cleanup(func() {
 		is.NoErr(i.Teardown(ctx))
 		is.NoErr(Cleanup(ctx, CleanupConfig{

source/logrepl/cleaner.go

@@ -31,7 +31,8 @@ type CleanupConfig struct {
 	PublicationName string
 }
 
-// Cleanup drops the provided replication and publication.
+// Cleanup drops the provided replication slot and publication.
+// It will terminate any backends consuming the replication slot before deletion.
 func Cleanup(ctx context.Context, c CleanupConfig) error {
 	pgconfig, err := pgconn.ParseConfig(c.URL)
 	if err != nil {
@@ -52,6 +53,14 @@ func Cleanup(ctx context.Context, c CleanupConfig) error {
 	var errs []error
 
 	if c.SlotName != "" {
+		// Terminate any outstanding backends which are consuming the slot before deleting it.
+		mrr := conn.Exec(ctx, fmt.Sprintf(
+			"SELECT pg_terminate_backend(active_pid) FROM pg_replication_slots WHERE slot_name='%s AND active=true'", c.SlotName,
+		))
+		if err := mrr.Close(); err != nil {
+			errs = append(errs, fmt.Errorf("failed to terminate active backends on slot: %w", err))
+		}
+
 		if err := pglogrepl.DropReplicationSlot(
 			ctx,
 			conn,