backup: add xlog adapter impl

.cspell_project-words.txt

@@ -37,6 +37,7 @@ heywoodlh
 iconfig
 incdir
 infof
+inprogress
 iproto
 leaderuuid
 libcluster
@@ -113,4 +114,5 @@ vshard
 vylog
 warnf
 xcall
+xdir
 xlog*

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/hashicorp/go-version v1.8.0
 	github.com/jedib0t/go-pretty/v6 v6.7.8
-	github.com/klauspost/compress v1.18.5
+	github.com/klauspost/compress v1.18.6
 	github.com/magefile/mage v1.15.0
 	github.com/manifoldco/promptui v0.9.0
 	github.com/mattn/go-isatty v0.0.20
@@ -30,10 +30,12 @@ require (
 	github.com/spf13/cobra v1.10.2
 	github.com/stretchr/testify v1.11.1
 	github.com/tarantool/cartridge-cli v0.0.0-00010101000000-000000000000
+	github.com/tarantool/go-iproto v1.1.0
 	github.com/tarantool/go-prompt v1.0.1
 	github.com/tarantool/go-storage v1.5.0
 	github.com/tarantool/go-tarantool v1.12.3
 	github.com/tarantool/go-tarantool/v2 v2.4.2
+	github.com/tarantool/go-xlog v0.0.0-20260707203858-fed522934686
 	github.com/tarantool/tt/lib/cluster v0.0.0
 	github.com/tarantool/tt/lib/connect v0.0.0-0
 	github.com/tarantool/tt/lib/dial v0.0.0-0
@@ -145,7 +147,6 @@ require (
 	github.com/soheilhy/cmux v0.1.5 // indirect
 	github.com/spacemonkeygo/spacelog v0.0.0-20180420211403-2296661a0572 // indirect
 	github.com/spf13/pflag v1.0.10 // indirect
-	github.com/tarantool/go-iproto v1.1.0 // indirect
 	github.com/tarantool/go-openssl v1.2.2 // indirect
 	github.com/tarantool/go-option v1.0.0 // indirect
 	github.com/tarantool/go-tlsdialer v1.0.2 // indirect

go.sum

@@ -180,8 +180,8 @@ github.com/jonboulle/clockwork v0.5.0/go.mod h1:3mZlmanh0g2NDKO5TWZVJAfofYk64M7X
 github.com/jpillora/backoff v0.0.0-20180909062703-3050d21c67d7/go.mod h1:2iMrUgbbvHEiQClaW2NsSzMyGHqN+rDFqY705q49KG0=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
-github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE=
-github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
+github.com/klauspost/compress v1.18.6 h1:2jupLlAwFm95+YDR+NwD2MEfFO9d4z4Prjl1XXDjuao=
+github.com/klauspost/compress v1.18.6/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
 github.com/klauspost/cpuid/v2 v2.0.1/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.11 h1:0OwqZRYI2rFrjS4kvkDnqJkKHdHaRnCm68/DY4OxRzU=
 github.com/klauspost/cpuid/v2 v2.2.11/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0=
@@ -369,6 +369,8 @@ github.com/tarantool/go-tarantool/v2 v2.4.2 h1:rkzYtFhLJLA9RDIhjzN93MJBN5PBxHW4+
 github.com/tarantool/go-tarantool/v2 v2.4.2/go.mod h1:MTbhdjFc3Jl63Lgi/UJr5D+QbT+QegqOzsNJGmaw7VM=
 github.com/tarantool/go-tlsdialer v1.0.2 h1:TiOkihvC2ufLbOqJcJLuQ9I7W5bsZtmnT7KHF/t8n4s=
 github.com/tarantool/go-tlsdialer v1.0.2/go.mod h1:ztE9J5oh4YzNQkSqlpjA0OkY89zHy/BiJJbbSFm+IiQ=
+github.com/tarantool/go-xlog v0.0.0-20260707203858-fed522934686 h1:2SjjYqDhtjen/3pHeSwpXDBM9dNZoXhuYb/s6vhKnU4=
+github.com/tarantool/go-xlog v0.0.0-20260707203858-fed522934686/go.mod h1:ZDu6simTStt+JG8BlHStzS2JAs/8k9HeynblSoR8+bo=
 github.com/testcontainers/testcontainers-go v0.43.0 h1:oEQx5MW2DGd9z3AeEQfB2lPM0eLs7ztyaGRu75bFo5A=
 github.com/testcontainers/testcontainers-go v0.43.0/go.mod h1:+VxkT2NQnKOZPKi6praMuMKYHYyOGXr0XSBSlSMCzFo=
 github.com/tinylib/msgp v1.6.1 h1:ESRv8eL3u+DNHUoSAAQRE50Hm162zqAnBoGv9PzScPY=

lib/backup/xlog/adapters.go

@@ -0,0 +1,204 @@
+package xlog
+
+import (
+	"errors"
+	"fmt"
+	"math"
+	"slices"
+
+	"github.com/google/uuid"
+
+	xdir "github.com/tarantool/go-xlog/dir"
+	"github.com/tarantool/go-xlog/filter"
+	"github.com/tarantool/go-xlog/format"
+	"github.com/tarantool/go-xlog/reader"
+	"github.com/tarantool/go-xlog/tools"
+	"github.com/tarantool/go-xlog/writer"
+
+	"github.com/tarantool/tt/lib/backup"
+)
+
+// PatchInstanceUUID rewrites the Instance field in the src header
+// writing to dst.
+func PatchInstanceUUID(src, dst, newUUID string) error {
+	id, err := uuid.Parse(newUUID)
+	if err != nil {
+		return fmt.Errorf("%w: %q: %w", ErrInvalidUUID, newUUID, err)
+	}
+
+	if src == dst {
+		err := tools.ReplaceInstanceUUIDInPlace(src, id)
+		if errors.Is(err, tools.ErrUUIDWidthMismatch) {
+			return fmt.Errorf("%w: %q", ErrInPlaceWidthMismatch, src)
+		}
+
+		if err != nil {
+			return fmt.Errorf("xlog: patch instance uuid in place %q: %w", src, err)
+		}
+
+		return nil
+	}
+
+	if err := tools.RewriteMeta(src, dst, tools.ReplaceInstanceUUID(id)); err != nil {
+		return fmt.Errorf("xlog: patch instance uuid %q to %q: %w", src, dst, err)
+	}
+
+	return nil
+}
+
+// TruncateAt copies transactions from src to dst up to and including the
+// recovery point, dropping the rest.
+func TruncateAt(src, dst string, replicaID uint32, lsn int64) error {
+	pred := filter.UntilVClock(format.VClock{replicaID: lsn})
+
+	outVClock, err := computeTruncatedVClock(src, pred)
+	if err != nil {
+		return err
+	}
+
+	meta, err := reader.ReadHeader(src)
+	if err != nil {
+		return fmt.Errorf("xlog: truncate: read header %q: %w", src, err)
+	}
+
+	meta.VClock = outVClock
+
+	return copyKeptTxs(src, dst, meta, pred)
+}
+
+// FindTrimFile returns the .xlog in dir whose transaction stream includes the
+// point (replicaID, lsn).
+func FindTrimFile(dir string, replicaID uint32, lsn int64) (string, error) {
+	d, err := xdir.OpenDir(dir, format.FiletypeXLOG)
+	if err != nil {
+		return "", fmt.Errorf("xlog: index xlog dir %q: %w", dir, err)
+	}
+
+	entry, err := d.LocateLSN(replicaID, lsn)
+	if err != nil {
+		if errors.Is(err, xdir.ErrNotFound) {
+			return "", fmt.Errorf("%w: dir %q replica %d lsn %d",
+				ErrTrimFileNotFound, dir, replicaID, lsn)
+		}
+
+		return "", fmt.Errorf("xlog: locate trim file in %q: %w", dir, err)
+	}
+
+	return entry.Path, nil
+}
+
+// toFormatVClock converts a manifest vclock to format.VClock.
+func toFormatVClock(v backup.Vclock) (format.VClock, error) {
+	if v == nil {
+		return nil, nil
+	}
+
+	out := make(format.VClock, len(v))
+
+	for id, lsn := range v {
+		if lsn > math.MaxInt64 {
+			return nil, fmt.Errorf("%w: replica %d lsn %d exceeds int64", ErrLSNOverflow, id, lsn)
+		}
+
+		out[id] = int64(lsn)
+	}
+
+	return out, nil
+}
+
+// fromFormatVClock is the inverse of toFormatVClock, returning ErrLSNOverflow
+// if any LSN is negative.
+func fromFormatVClock(v format.VClock) (backup.Vclock, error) {
+	if v == nil {
+		return nil, nil
+	}
+
+	out := make(backup.Vclock, len(v))
+
+	for id, lsn := range v {
+		if lsn < 0 {
+			return nil, fmt.Errorf("%w: replica %d lsn %d is negative", ErrLSNOverflow, id, lsn)
+		}
+
+		out[id] = uint64(lsn)
+	}
+
+	return out, nil
+}
+
+// computeTruncatedVClock returns the per-replica high-water of every row in
+// every kept transaction — the exact VClock for the truncated output.
+func computeTruncatedVClock(src string, pred filter.Filter) (format.VClock, error) {
+	r, err := reader.Open(src)
+	if err != nil {
+		return nil, fmt.Errorf("xlog: truncate: open %q: %w", src, err)
+	}
+
+	defer func() { _ = r.Close() }()
+
+	out := format.VClock{}
+
+	for tx, err := range r.Txs() {
+		if err != nil {
+			return nil, fmt.Errorf("xlog: truncate: read tx from %q: %w", src, err)
+		}
+
+		if !slices.ContainsFunc(tx.Rows, pred) {
+			continue
+		}
+
+		for i := range tx.Rows {
+			if row := tx.Rows[i]; row.LSN > out[row.ReplicaID] {
+				out[row.ReplicaID] = row.LSN
+			}
+		}
+	}
+
+	return out, nil
+}
+
+// copyKeptTxs writes meta to dst and streams every transaction from src that
+// passes pred, discarding the .inprogress file on any error.
+func copyKeptTxs(src, dst string, meta *format.Meta, pred filter.Filter) error {
+	r, err := reader.Open(src)
+	if err != nil {
+		return fmt.Errorf("xlog: truncate: open %q: %w", src, err)
+	}
+
+	defer func() { _ = r.Close() }()
+
+	w, err := writer.Create(dst, meta)
+	if err != nil {
+		return fmt.Errorf("xlog: truncate: create %q: %w", dst, err)
+	}
+
+	committed := false
+
+	defer func() {
+		if !committed {
+			_ = w.Discard()
+		}
+	}()
+
+	for tx, err := range r.Txs() {
+		if err != nil {
+			return fmt.Errorf("xlog: truncate: read tx from %q: %w", src, err)
+		}
+
+		if !slices.ContainsFunc(tx.Rows, pred) {
+			continue
+		}
+
+		if err := w.WriteTx(tx.Rows); err != nil {
+			return fmt.Errorf("xlog: truncate: write tx to %q: %w", dst, err)
+		}
+	}
+
+	if err := w.Close(); err != nil {
+		return fmt.Errorf("xlog: truncate: finalize %q: %w", dst, err)
+	}
+
+	committed = true
+
+	return nil
+}

lib/backup/xlog/convert_test.go

@@ -0,0 +1,86 @@
+package xlog
+
+import (
+	"errors"
+	"math"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/tarantool/go-xlog/format"
+
+	"github.com/tarantool/tt/lib/backup"
+)
+
+func TestToFormatVClock_RoundTrip(t *testing.T) {
+	src := backup.Vclock{0: 0, 1: 42, 2: 9000}
+
+	got, err := toFormatVClock(src)
+	require.NoError(t, err)
+	require.Equal(t, format.VClock{0: 0, 1: 42, 2: 9000}, got)
+
+	back, err := fromFormatVClock(got)
+	require.NoError(t, err)
+	require.Equal(t, src, back)
+}
+
+func TestToFormatVClock_Empty(t *testing.T) {
+	got, err := toFormatVClock(backup.Vclock{})
+	require.NoError(t, err)
+	require.Equal(t, format.VClock{}, got)
+}
+
+func TestToFormatVClock_Nil(t *testing.T) {
+	got, err := toFormatVClock(nil)
+	require.NoError(t, err)
+	require.Empty(t, got)
+}
+
+func TestToFormatVClock_Overflow(t *testing.T) {
+	src := backup.Vclock{1: math.MaxInt64 + 1}
+
+	_, err := toFormatVClock(src)
+	require.ErrorIs(t, err, ErrLSNOverflow)
+}
+
+func TestToFormatVClock_MaxInt64OK(t *testing.T) {
+	src := backup.Vclock{1: math.MaxInt64}
+
+	got, err := toFormatVClock(src)
+	require.NoError(t, err)
+	require.Equal(t, int64(math.MaxInt64), got[1])
+}
+
+func TestFromFormatVClock_Negative(t *testing.T) {
+	src := format.VClock{1: -1}
+
+	_, err := fromFormatVClock(src)
+	require.ErrorIs(t, err, ErrLSNOverflow)
+}
+
+func TestFromFormatVClock_Nil(t *testing.T) {
+	got, err := fromFormatVClock(nil)
+	require.NoError(t, err)
+	require.Empty(t, got)
+}
+
+func TestVClock_RoundTripBoundaries(t *testing.T) {
+	src := backup.Vclock{
+		0:  0,
+		1:  1,
+		7:  math.MaxInt64,
+		13: 1 << 40,
+	}
+
+	fv, err := toFormatVClock(src)
+	require.NoError(t, err)
+
+	back, err := fromFormatVClock(fv)
+	require.NoError(t, err)
+	require.Equal(t, src, back)
+}
+
+func TestVClock_OverflowIsSentinel(t *testing.T) {
+	_, err := toFormatVClock(backup.Vclock{1: math.MaxUint64})
+	require.True(t, errors.Is(err, ErrLSNOverflow))
+}

lib/backup/xlog/errors.go

@@ -0,0 +1,11 @@
+package xlog
+
+import "errors"
+
+var (
+	ErrInvalidUUID          = errors.New("xlog: invalid instance UUID")
+	ErrTrimFileNotFound     = errors.New("xlog: no xlog file contains the recovery point")
+	ErrLSNOverflow          = errors.New("xlog: lsn out of range for int64<->uint64 conversion")
+	ErrInPlaceWidthMismatch = errors.New(
+		"xlog: on-disk UUID width differs from replacement; use distinct src and dst")
+)