cdc: bound catchup segment work

go.mod

@@ -2,7 +2,7 @@ module github.com/GetStream/pgmigrate
 
 go 1.25.0
 
-toolchain go1.25.12
+toolchain go1.25.13
 
 require (
 	github.com/jackc/pglogrepl v0.0.0-20260401131349-e37c41485510

internal/app/app.go

@@ -605,7 +605,9 @@ func (a App) Run(ctx context.Context, cfg config.Config) (runErr error) {
 		if err := pauseForCrashTest(groupCtx, state.PhaseCatchup); err != nil {
 			return err
 		}
-		return runApplierToFollow(groupCtx, cfg, store, holder.Snapshot, durable, state.PhaseCatchup)
+		return runApplierToFollow(
+			groupCtx, cfg, store, holder.Snapshot, durable, writer.SegmentCatalog(), state.PhaseCatchup,
+		)
 	})
 	if cfg.Metrics != "" {
 		group.Go(func() error { return serveMetrics(groupCtx, cfg.Metrics, store) })
@@ -708,7 +710,9 @@ func (a App) resumePostCopy(ctx context.Context, cfg config.Config, store *state
 				return err
 			}
 		}
-		return runApplierToFollow(groupCtx, cfg, store, snapshot, durable, phase)
+		return runApplierToFollow(
+			groupCtx, cfg, store, snapshot, durable, writer.SegmentCatalog(), phase,
+		)
 	})
 	group.Go(func() error {
 		ticker := time.NewTicker(200 * time.Millisecond)
@@ -1329,6 +1333,7 @@ func runApplierToFollow(
 	store *state.Store,
 	snapshot setup.Snapshot,
 	durable *cdc.DurableWatermark,
+	segments *cdc.SegmentCatalog,
 	phase state.Phase,
 ) error {
 	migration, err := store.Migration(ctx)
@@ -1338,6 +1343,7 @@ func runApplierToFollow(
 	pruner, err := cdc.NewSegmentPruner(cdc.SegmentPrunerConfig{
 		Directory: filepath.Join(cfg.Dir, "cdc"),
 		Interval:  cfg.SegmentPruneInterval,
+		Catalog:   segments,
 	})
 	if err != nil {
 		return err
@@ -1365,13 +1371,13 @@ func runApplierToFollow(
 		return err
 	}
 	if phase == state.PhaseFollow || phase == state.PhaseDrained || phase == state.PhaseCutover {
-		return runApplierContinuous(ctx, applier, store)
+		return runApplierWithPruner(ctx, applier, pruner, store)
 	}
 	boundary := durable.Load()
 	applyCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	result := make(chan error, 1)
-	go func() { result <- runApplierContinuous(applyCtx, applier, store) }()
+	go func() { result <- runApplierWithPruner(applyCtx, applier, pruner, store) }()
 	if err := awaitCatchup(ctx, boundary, applier.WaitUntil, result); err != nil {
 		return err
 	}
@@ -1390,6 +1396,18 @@ func runApplierToFollow(
 	}
 }
 
+func runApplierWithPruner(
+	ctx context.Context,
+	applier *cdc.Applier,
+	pruner *cdc.SegmentPruner,
+	store *state.Store,
+) error {
+	group, groupCtx := errgroup.WithContext(ctx)
+	group.Go(func() error { return pruner.Run(groupCtx) })
+	group.Go(func() error { return runApplierContinuous(groupCtx, applier, store) })
+	return group.Wait()
+}
+
 // awaitCatchup waits for target apply progress to reach boundary while watching
 // the applier that produces it. The wait polls a row on the target that only the
 // applier advances, so an applier exit has to end the wait as well: reading the

internal/cdc/applier.go

@@ -110,14 +110,16 @@ func (a *Applier) Run(ctx context.Context) error {
 // WaitUntil blocks until the authoritative target progress reaches boundary.
 // Only Applier.Run advances that progress, so the caller must supervise Run
 // concurrently and stop waiting if it exits; otherwise this polls forever.
+//
+// boundary must already be a transaction EndLSN. Catchup passes the durable
+// watermark, which the persister published from a commit. A manual cutover LSN
+// is normalized before it is stored. This wait must not scan the segment
+// directory: NormalizeEndPosition reads every staged transaction, and after a
+// long copy that is the whole backlog.
 func (a *Applier) WaitUntil(ctx context.Context, boundary LSN) error {
 	if boundary == 0 {
 		return nil
 	}
-	// Resolve the boundary at most once. It cannot move afterwards, and
-	// resolving it per poll re-decoded the whole staged stream each time.
-	effectiveBoundary := boundary
-	resolved := false
 	for {
 		conn, err := postgres.Connect(ctx, a.config.ConnString)
 		if err != nil {
@@ -128,17 +130,7 @@ func (a *Applier) WaitUntil(ctx context.Context, boundary LSN) error {
 		if readErr != nil {
 			return fmt.Errorf("cdc: read catch-up progress: %w", readErr)
 		}
-		if !resolved {
-			if durable := a.config.Durable.Load(); durable >= boundary {
-				resolution, err := NormalizeEndPosition(a.config.Directory, boundary, durable)
-				if err != nil {
-					return err
-				}
-				effectiveBoundary = resolution.Boundary
-				resolved = true
-			}
-		}
-		if LSN(progress) >= effectiveBoundary {
+		if LSN(progress) >= boundary {
 			return nil
 		}
 		timer := time.NewTimer(a.config.PollInterval)

internal/cdc/cdc_integration_test.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"path/filepath"
 	"slices"
 	"strings"
 	"sync"
@@ -224,6 +225,7 @@ func TestPG17LiveWALStageApplyCrashRetry(t *testing.T) {
 	pruner, err := NewSegmentPruner(SegmentPrunerConfig{
 		Directory: directory,
 		Interval:  time.Nanosecond,
+		Catalog:   writer.SegmentCatalog(),
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -460,6 +462,176 @@ func (c *collectingSampler) all() []string {
 	return slices.Compact(seen)
 }
 
+// TestPG17WaitUntilDoesNotScanStagedSegments is the catchup wait: the boundary
+// is already the durable EndLSN. Scanning the segment directory to "normalize"
+// it would decode the whole backlog after a long copy, which is how a shard
+// sat on the first file at the memory ceiling with apply idle.
+func TestPG17WaitUntilDoesNotScanStagedSegments(t *testing.T) {
+	target := pgtest.Start(t, 17)
+	ctx := context.Background()
+	conn := target.Connect(t)
+	if err := postgres.EnsureProgressTable(ctx, conn); err != nil {
+		t.Fatal(err)
+	}
+	if err := postgres.UpdateProgress(ctx, conn, "catchup", 0x100); err != nil {
+		t.Fatal(err)
+	}
+	watermark := new(DurableWatermark)
+	watermark.Publish(0x100)
+	applier, err := NewApplier(ApplierConfig{
+		ConnString:   target.URI,
+		Directory:    filepath.Join(t.TempDir(), "empty-cdc"),
+		StreamID:     "catchup",
+		Durable:      watermark,
+		PollInterval: 5 * time.Millisecond,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitCtx, cancel := context.WithTimeout(ctx, 15*time.Second)
+	defer cancel()
+	if err := applier.WaitUntil(waitCtx, 0x100); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestPG17ApplierStartsBeforeReadingUnappliedSuffix(t *testing.T) {
+	target := pgtest.Start(t, 17)
+	ctx := context.Background()
+	targetSQL := target.Connect(t)
+	if _, err := targetSQL.Exec(ctx, `
+		CREATE TABLE public.catchup_probe (
+			id bigint PRIMARY KEY,
+			value text NOT NULL
+		)`); err != nil {
+		t.Fatal(err)
+	}
+
+	directory := t.TempDir()
+	writer, _, err := OpenWriter(WriterConfig{Directory: directory, RotationBytes: 1})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var lastEnd LSN
+	for i := 1; i <= 256; i++ {
+		value := fmt.Sprint(i)
+		row := Tuple{
+			{Kind: DatumText, Data: []byte(value)},
+			{Kind: DatumText, Data: []byte("value-" + value)},
+		}
+		transaction := Transaction{
+			CommitLSN:  LSN(i * 0x10),
+			EndLSN:     LSN(i*0x10 + 1),
+			CommitTime: time.Unix(int64(i), 0).UTC(),
+			Relations: []Relation{{
+				OID:             4242,
+				Namespace:       "public",
+				Name:            "catchup_probe",
+				ReplicaIdentity: 'd',
+				Columns: []Column{
+					{Name: "id", Type: 20, Flags: 1},
+					{Name: "value", Type: 25},
+				},
+			}},
+			Changes: []Change{{
+				RelationOID: 4242,
+				Kind:        ChangeInsert,
+				New:         &row,
+			}},
+		}
+		lastEnd = transaction.EndLSN
+		if err := writer.Append(&transaction); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := writer.Close(); err != nil {
+		t.Fatal(err)
+	}
+	ranges := writer.SegmentCatalog().snapshot()
+	last := ranges[len(ranges)-1]
+	file, err := os.OpenFile(last.Path, os.O_WRONLY, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := file.WriteAt([]byte{0xff}, frameHeaderSize); err != nil {
+		_ = file.Close()
+		t.Fatal(err)
+	}
+	if err := file.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	const streamID = "catchup-starts-before-suffix"
+	const generation = "generation-1"
+	if err := EnsureStreamProgressIdentity(ctx, targetSQL, StreamIdentityConfig{
+		StreamID: streamID, Generation: generation, FreshSetup: true,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if err := postgres.UpdateProgress(ctx, targetSQL, streamID, 0); err != nil {
+		t.Fatal(err)
+	}
+	if err := EnsureStreamProgressIdentity(ctx, targetSQL, StreamIdentityConfig{
+		StreamID: streamID, Generation: generation, FreshSetup: true,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	watermark := new(DurableWatermark)
+	watermark.Publish(lastEnd)
+	pruner, err := NewSegmentPruner(SegmentPrunerConfig{
+		Directory: directory,
+		Interval:  time.Nanosecond,
+		Catalog:   writer.SegmentCatalog(),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	applier, err := NewApplier(ApplierConfig{
+		ConnString:           target.URI,
+		Directory:            directory,
+		StreamID:             streamID,
+		StreamGeneration:     generation,
+		TargetHasCopiedData:  true,
+		Durable:              watermark,
+		PollInterval:         time.Millisecond,
+		AfterProgress:        pruner.OnProgress,
+		ReaderSpillDirectory: filepath.Join(t.TempDir(), "reader-spill"),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	applyCtx, stop := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- applier.Run(applyCtx) }()
+	deadline := time.Now().Add(30 * time.Second)
+	var progress pglogrepl.LSN
+	for time.Now().Before(deadline) {
+		progress, _, err = postgres.ReadProgress(ctx, targetSQL, streamID)
+		if err != nil {
+			stop()
+			t.Fatal(err)
+		}
+		if progress > 0 {
+			break
+		}
+		select {
+		case err := <-done:
+			stop()
+			t.Fatalf("applier reached corrupt suffix before first progress: %v", err)
+		default:
+		}
+		time.Sleep(time.Millisecond)
+	}
+	if progress == 0 {
+		stop()
+		t.Fatal("target progress did not move before the unapplied suffix")
+	}
+	stop()
+	if err := <-done; err != nil && !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+}
+
 func waitFor(t testing.TB, timeout time.Duration, condition func() bool) {
 	t.Helper()
 	deadline := time.Now().Add(timeout)