Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Retry when WAL segment has not been found #930

Merged
merged 1 commit into from
Dec 29, 2023
Merged

Retry when WAL segment has not been found #930

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions flow/activities/flowable.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -271,7 +271,7 @@ func (a *FlowableActivity) StartFlow(ctx context.Context,
err = errGroup.Wait()
if err != nil {
a.Alerter.LogFlowError(ctx, flowName, err)
return nil, fmt.Errorf("failed to pull records: %w", err)
return nil, fmt.Errorf("failed in pull records when: %w", err)
}
slog.InfoContext(ctx, "no records to push")
syncResponse := &model.SyncResponse{}
Expand Down Expand Up @@ -615,7 +615,7 @@ func (a *FlowableActivity) replicateQRepPartition(ctx context.Context,
recordBatch, err := srcConn.PullQRepRecords(config, partition)
if err != nil {
a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
return fmt.Errorf("failed to pull records: %w", err)
return fmt.Errorf("failed to pull qrep records: %w", err)
}
numRecords := int64(recordBatch.NumRecords)
slog.InfoContext(ctx, fmt.Sprintf("pulled %d records\n", len(recordBatch.Records)))
Expand Down Expand Up @@ -955,7 +955,7 @@ func (a *FlowableActivity) ReplicateXminPartition(ctx context.Context,
numRecords, currentSnapshotXmin, pullErr = pgConn.PullXminRecordStream(config, partition, stream)
if pullErr != nil {
a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
slog.InfoContext(ctx, fmt.Sprintf("failed to pull records: %v", err))
slog.InfoContext(ctx, fmt.Sprintf("[xmin] failed to pull records: %v", err))
return err
}

Expand Down
90 changes: 68 additions & 22 deletions flow/connectors/postgres/cdc.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (
"encoding/json"
"fmt"
"log/slog"
"regexp"
"time"

"github.com/PeerDB-io/peer-flow/connectors/utils"
Expand All @@ -24,6 +25,8 @@ import (
"go.temporal.io/sdk/activity"
)

const maxRetriesForWalSegmentRemoved = 5

type PostgresCDCSource struct {
ctx context.Context
replPool *pgxpool.Pool
Expand All @@ -44,6 +47,8 @@ type PostgresCDCSource struct {
// for storing chema delta audit logs to catalog
catalogPool *pgxpool.Pool
flowJobName string

walSegmentRemovedRegex *regexp.Regexp
}

type PostgresCDCConfig struct {
Expand All @@ -59,13 +64,22 @@ type PostgresCDCConfig struct {
SetLastOffset func(int64) error
}

type startReplicationOpts struct {
conn *pgconn.PgConn
startLSN pglogrepl.LSN
replicationOpts pglogrepl.StartReplicationOptions
}

// Create a new PostgresCDCSource
func NewPostgresCDCSource(cdcConfig *PostgresCDCConfig, customTypeMap map[uint32]string) (*PostgresCDCSource, error) {
childToParentRelIDMap, err := getChildToParentRelIDMap(cdcConfig.AppContext, cdcConfig.Connection)
if err != nil {
return nil, fmt.Errorf("error getting child to parent relid map: %w", err)
}

pattern := "requested WAL segment .* has already been removed.*"
regex := regexp.MustCompile(pattern)

flowName, _ := cdcConfig.AppContext.Value(shared.FlowNameKey).(string)
return &PostgresCDCSource{
ctx: cdcConfig.AppContext,
Expand All @@ -83,6 +97,7 @@ func NewPostgresCDCSource(cdcConfig *PostgresCDCConfig, customTypeMap map[uint32
logger: *slog.With(slog.String(string(shared.FlowNameKey), flowName)),
catalogPool: cdcConfig.CatalogPool,
flowJobName: cdcConfig.FlowJobName,
walSegmentRemovedRegex: regex,
}, nil
}

Expand Down Expand Up @@ -120,21 +135,11 @@ func getChildToParentRelIDMap(ctx context.Context, pool *pgxpool.Pool) (map[uint

// PullRecords pulls records from the cdc stream
func (p *PostgresCDCSource) PullRecords(req *model.PullRecordsRequest) error {
// setup options
pluginArguments := []string{
"proto_version '1'",
}

if p.publication != "" {
pubOpt := fmt.Sprintf("publication_names '%s'", p.publication)
pluginArguments = append(pluginArguments, pubOpt)
} else {
return fmt.Errorf("publication name is not set")
replicationOpts, err := p.replicationOptions()
if err != nil {
return fmt.Errorf("error getting replication options: %w", err)
}

replicationOpts := pglogrepl.StartReplicationOptions{PluginArgs: pluginArguments}
replicationSlot := p.slot

// create replication connection
replicationConn, err := p.replPool.Acquire(p.ctx)
if err != nil {
Expand All @@ -146,13 +151,6 @@ func (p *PostgresCDCSource) PullRecords(req *model.PullRecordsRequest) error {
pgConn := replicationConn.Conn().PgConn()
p.logger.Info("created replication connection")

sysident, err := pglogrepl.IdentifySystem(p.ctx, pgConn)
if err != nil {
return fmt.Errorf("IdentifySystem failed: %w", err)
}
p.logger.Debug(fmt.Sprintf("SystemID: %s, Timeline: %d, XLogPos: %d, DBName: %s",
sysident.SystemID, sysident.Timeline, sysident.XLogPos, sysident.DBName))

// start replication
var clientXLogPos, startLSN pglogrepl.LSN
if req.LastOffset > 0 {
Expand All @@ -161,15 +159,48 @@ func (p *PostgresCDCSource) PullRecords(req *model.PullRecordsRequest) error {
startLSN = clientXLogPos + 1
}

err = pglogrepl.StartReplication(p.ctx, pgConn, replicationSlot, startLSN, replicationOpts)
opts := startReplicationOpts{
conn: pgConn,
startLSN: startLSN,
replicationOpts: *replicationOpts,
}

err = p.startReplication(opts)
if err != nil {
return fmt.Errorf("error starting replication at startLsn - %d: %w", startLSN, err)
return fmt.Errorf("error starting replication: %w", err)
}

p.logger.Info(fmt.Sprintf("started replication on slot %s at startLSN: %d", p.slot, startLSN))

return p.consumeStream(pgConn, req, clientXLogPos, req.RecordStream)
}

func (p *PostgresCDCSource) startReplication(opts startReplicationOpts) error {
err := pglogrepl.StartReplication(p.ctx, opts.conn, p.slot, opts.startLSN, opts.replicationOpts)
if err != nil {
p.logger.Error("error starting replication", slog.Any("error", err))
return fmt.Errorf("error starting replication at startLsn - %d: %w", opts.startLSN, err)
}

p.logger.Info(fmt.Sprintf("started replication on slot %s at startLSN: %d", p.slot, opts.startLSN))
return nil
}

func (p *PostgresCDCSource) replicationOptions() (*pglogrepl.StartReplicationOptions, error) {
pluginArguments := []string{
"proto_version '1'",
}

if p.publication != "" {
pubOpt := fmt.Sprintf("publication_names '%s'", p.publication)
pluginArguments = append(pluginArguments, pubOpt)
} else {
return nil, fmt.Errorf("publication name is not set")
}

return &pglogrepl.StartReplicationOptions{PluginArgs: pluginArguments}, nil
}

// start consuming the cdc stream
func (p *PostgresCDCSource) consumeStream(
conn *pgconn.PgConn,
Expand Down Expand Up @@ -250,6 +281,7 @@ func (p *PostgresCDCSource) consumeStream(

pkmRequiresResponse := false
waitingForCommit := false
retryAttemptForWALSegmentRemoved := 0

for {
if pkmRequiresResponse {
Expand Down Expand Up @@ -324,6 +356,20 @@ func (p *PostgresCDCSource) consumeStream(
}

if errMsg, ok := rawMsg.(*pgproto3.ErrorResponse); ok {
if errMsg.Severity == "ERROR" && errMsg.Code == "XX000" {
serprex marked this conversation as resolved.
Show resolved Hide resolved
if p.walSegmentRemovedRegex.MatchString(errMsg.Message) {
retryAttemptForWALSegmentRemoved++
if retryAttemptForWALSegmentRemoved > maxRetriesForWalSegmentRemoved {
return fmt.Errorf("max retries for WAL segment removed exceeded: %+v", errMsg)
} else {
p.logger.Warn(fmt.Sprintf(
"WAL segment removed, restarting replication retrying in 30 seconds..."),
slog.Any("error", errMsg), slog.Int("retryAttempt", retryAttemptForWALSegmentRemoved))
time.Sleep(30 * time.Second)
continue
}
}
}
return fmt.Errorf("received Postgres WAL error: %+v", errMsg)
}

Expand Down
Loading