fix(gtc): prevent CDC data loss on poison records, streamed transactions, and reconnects

services/gtc/internal/adapters/primary/wal/decoder.go

@@ -1,6 +1,7 @@
 package wal
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/emoss08/gtc/internal/core/domain"
@@ -12,7 +13,6 @@ import (
 type Decoder struct {
 	relations          map[uint32]*pglogrepl.RelationMessageV2
 	typeMap            *pgtype.Map
-	inStream           bool
 	currentTransaction transactionState
 }
 
@@ -65,8 +65,13 @@ func (d *Decoder) Decode(rawMsg pgproto3.BackendMessage) (*DecodeResult, error)
 	return &DecodeResult{}, nil
 }
 
+func (d *Decoder) Reset() {
+	d.relations = make(map[uint32]*pglogrepl.RelationMessageV2)
+	d.currentTransaction = transactionState{}
+}
+
 func (d *Decoder) decodeWALData(walData []byte, lsn pglogrepl.LSN) (*domain.TransactionRecords, error) {
-	logicalMsg, err := pglogrepl.ParseV2(walData, d.inStream)
+	logicalMsg, err := pglogrepl.ParseV2(walData, false)
 	if err != nil {
 		return nil, err
 	}
@@ -151,11 +156,14 @@ func (d *Decoder) decodeWALData(walData []byte, lsn pglogrepl.LSN) (*domain.Tran
 			}
 		}
 
-	case *pglogrepl.StreamStartMessageV2:
-		d.inStream = true
-
-	case *pglogrepl.StreamStopMessageV2:
-		d.inStream = false
+	case *pglogrepl.StreamStartMessageV2,
+		*pglogrepl.StreamStopMessageV2,
+		*pglogrepl.StreamCommitMessageV2,
+		*pglogrepl.StreamAbortMessageV2:
+		return nil, fmt.Errorf(
+			"received streamed transaction message %T: in-progress transaction streaming is not supported and must stay disabled in the replication plugin arguments",
+			msg,
+		)
 	}
 
 	return nil, nil

services/gtc/internal/adapters/primary/wal/decoder_test.go

@@ -50,6 +50,54 @@ func TestDecoderEmitsCommittedTransaction(t *testing.T) {
 	}
 }
 
+func TestDecoderRejectsStreamedTransactionMessages(t *testing.T) {
+	t.Parallel()
+
+	decoder := NewDecoder()
+
+	if _, err := decoder.decodeWALData(encodeStreamStartMessage(42), pglogrepl.LSN(90)); err == nil {
+		t.Fatalf("expected streamed transaction message to be rejected")
+	}
+}
+
+func TestDecoderResetClearsTransactionState(t *testing.T) {
+	t.Parallel()
+
+	decoder := NewDecoder()
+	beginTime := time.Date(2026, 3, 20, 20, 12, 34, 0, time.UTC)
+
+	if _, err := decoder.decodeWALData(encodeBeginMessage(pglogrepl.LSN(100), beginTime, 42), pglogrepl.LSN(90)); err != nil {
+		t.Fatalf("decode begin: %v", err)
+	}
+	decoder.appendRecord(domain.SourceRecord{
+		Operation: domain.OperationInsert,
+		Schema:    "public",
+		Table:     "shipments",
+		NewData:   map[string]any{"id": "shp_1"},
+	})
+	decoder.relations[7] = &pglogrepl.RelationMessageV2{}
+
+	decoder.Reset()
+
+	if len(decoder.relations) != 0 {
+		t.Fatalf("expected relations to be cleared, got %d", len(decoder.relations))
+	}
+	if decoder.currentTransaction.records != nil {
+		t.Fatalf("expected buffered transaction records to be cleared")
+	}
+	if decoder.currentTransaction.xid != 0 {
+		t.Fatalf("expected transaction xid to be cleared, got %d", decoder.currentTransaction.xid)
+	}
+}
+
+func encodeStreamStartMessage(xid uint32) []byte {
+	buf := make([]byte, 1+4+1)
+	buf[0] = byte(pglogrepl.MessageTypeStreamStart)
+	binary.BigEndian.PutUint32(buf[1:5], xid)
+	buf[5] = 1
+	return buf
+}
+
 func encodeBeginMessage(finalLSN pglogrepl.LSN, commitTime time.Time, xid uint32) []byte {
 	buf := make([]byte, 1+8+8+4)
 	buf[0] = byte(pglogrepl.MessageTypeBegin)

services/gtc/internal/adapters/primary/wal/reader.go

@@ -111,7 +111,7 @@ func (r *Reader) Start(ctx context.Context, startLSN string, handler ports.Trans
 			continue
 		}
 
-		if err := r.setupReplication(ctx, startLSN); err != nil {
+		if err := r.setupReplication(ctx, r.resumeLSN(startLSN)); err != nil {
 			r.logger.Error("replication setup failed", zap.Error(err))
 			r.stopSlotMonitor()
 			r.closeConnection(ctx)
@@ -217,6 +217,7 @@ func (r *Reader) setupReplication(ctx context.Context, startLSN string) error {
 		return err
 	}
 
+	r.decoder.Reset()
 	r.clientLSN.Store(uint64(effectiveLSN))
 	r.startSlotMonitor(ctx)
 	r.logger.Info("replication started", zap.String("lsn", effectiveLSN.String()))
@@ -575,7 +576,7 @@ func (r *Reader) startReplicationWithRetry(ctx context.Context, startLSN pglogre
 		"proto_version '2'",
 		fmt.Sprintf("publication_names '%s'", r.config.PublicationName),
 		"messages 'true'",
-		"streaming 'true'",
+		"streaming 'false'",
 	}
 
 	deadline := time.Now().Add(r.config.SlotRetryTimeout)
@@ -689,6 +690,14 @@ func (r *Reader) streamLoop(ctx context.Context, handler ports.TransactionHandle
 	}
 }
 
+func (r *Reader) resumeLSN(startLSN string) string {
+	if current := r.clientLSN.Load(); current != 0 {
+		return pglogrepl.LSN(current).String()
+	}
+
+	return startLSN
+}
+
 func (r *Reader) AdvanceLSN(lsn string) error {
 	parsedLSN, err := pglogrepl.ParseLSN(lsn)
 	if err != nil {

services/gtc/internal/adapters/primary/wal/reader_test.go

@@ -52,6 +52,28 @@ func TestAdvanceLSNDoesNotMoveBackward(t *testing.T) {
 	}
 }
 
+func TestResumeLSNUsesStartLSNBeforeFirstAdvance(t *testing.T) {
+	t.Parallel()
+
+	reader := &Reader{}
+	if got := reader.resumeLSN("0/10"); got != "0/10" {
+		t.Fatalf("expected resume lsn 0/10 before any advance, got %s", got)
+	}
+}
+
+func TestResumeLSNPrefersAdvancedLSN(t *testing.T) {
+	t.Parallel()
+
+	reader := &Reader{}
+	if err := reader.AdvanceLSN("0/20"); err != nil {
+		t.Fatalf("AdvanceLSN returned error: %v", err)
+	}
+
+	if got := reader.resumeLSN("0/10"); got != "0/20" {
+		t.Fatalf("expected resume lsn to use the advanced position 0/20, got %s", got)
+	}
+}
+
 func TestAdvanceLSNRejectsInvalidValue(t *testing.T) {
 	t.Parallel()
 

services/gtc/internal/core/services/runtime.go

@@ -348,14 +348,61 @@ func (r *Runtime) handleRecordWithProjections(
 			zap.String("sink", string(projection.Destination.Kind)),
 		)
 
-		if err := r.writeProjection(ctx, projection, record); err != nil {
+		if err := r.writeProjectionOrDeadLetter(ctx, projection, record); err != nil {
 			return err
 		}
 	}
 
 	return nil
 }
 
+func (r *Runtime) writeProjectionOrDeadLetter(
+	ctx context.Context,
+	projection domain.Projection,
+	record domain.SourceRecord,
+) error {
+	writeErr := r.writeProjection(ctx, projection, record)
+	if writeErr == nil {
+		return nil
+	}
+	if ctx.Err() != nil {
+		return writeErr
+	}
+	if r.dlqWriter == nil {
+		return writeErr
+	}
+
+	entry := domain.DeadLetterRecord{
+		TransactionID: record.Metadata.TransactionID,
+		CommitLSN:     record.Metadata.CommitLSN,
+		Projection:    projection.Name,
+		Error:         writeErr.Error(),
+		Attempts:      r.retryMax,
+		Record:        record,
+		CreatedAt:     time.Now().UTC(),
+	}
+	if dlqErr := r.dlqWriter.Write(ctx, entry); dlqErr != nil {
+		return fmt.Errorf(
+			"projection %s failed (%s) and dlq write failed: %w",
+			projection.Name,
+			writeErr.Error(),
+			dlqErr,
+		)
+	}
+
+	metrics.DeadLetteredRecords.WithLabelValues(projection.Name).Inc()
+	r.logger.Error("projection record sent to dlq, continuing",
+		zap.String("projection", projection.Name),
+		zap.String("table", record.FullTableName()),
+		zap.String("operation", record.Operation.String()),
+		zap.String("commit_lsn", record.Metadata.CommitLSN),
+		zap.Uint32("transaction_id", record.Metadata.TransactionID),
+		zap.Error(writeErr),
+	)
+
+	return nil
+}
+
 func (r *Runtime) writeProjection(ctx context.Context, projection domain.Projection, record domain.SourceRecord) error {
 	sink, ok := r.sinks[projection.Destination.Kind]
 	if !ok {
@@ -374,6 +421,7 @@ func (r *Runtime) writeProjection(ctx context.Context, projection domain.Project
 
 		lastErr = err
 		r.setStatus(sink.Name(), false)
+		metrics.SinkErrors.WithLabelValues(sink.Name(), "write").Inc()
 		r.logger.Warn("projection write failed",
 			zap.String("projection", projection.Name),
 			zap.String("sink", sink.Name()),
@@ -386,6 +434,7 @@ func (r *Runtime) writeProjection(ctx context.Context, projection domain.Project
 		if attempt == r.retryMax {
 			break
 		}
+		metrics.RetryAttempts.WithLabelValues(sink.Name()).Inc()
 
 		select {
 		case <-ctx.Done():
@@ -394,29 +443,6 @@ func (r *Runtime) writeProjection(ctx context.Context, projection domain.Project
 		}
 	}
 
-	if r.dlqWriter != nil {
-		entry := domain.DeadLetterRecord{
-			TransactionID: record.Metadata.TransactionID,
-			CommitLSN:     record.Metadata.CommitLSN,
-			Projection:    projection.Name,
-			Error:         lastErr.Error(),
-			Attempts:      r.retryMax,
-			Record:        record,
-			CreatedAt:     time.Now().UTC(),
-		}
-		if err := r.dlqWriter.Write(ctx, entry); err != nil {
-			return fmt.Errorf("projection %s failed and dlq write failed: %w", projection.Name, err)
-		}
-		r.logger.Error("projection sent to dlq",
-			zap.String("projection", projection.Name),
-			zap.String("table", record.FullTableName()),
-			zap.String("operation", record.Operation.String()),
-			zap.String("commit_lsn", record.Metadata.CommitLSN),
-			zap.Uint32("transaction_id", record.Metadata.TransactionID),
-			zap.Error(lastErr),
-		)
-	}
-
 	return fmt.Errorf("projection %s failed after %d attempts: %w", projection.Name, r.retryMax, lastErr)
 }