Skip to content

fix(replica): snapshot missing local L0 gaps #1463

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

Draft
corylanou wants to merge 1 commit into
base: issue-1457-gap-heal-upload-only-missing
Choose a base branch
from
Draft

fix(replica): snapshot missing local L0 gaps #1463

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
45 changes: 39 additions & 6 deletions replica.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,7 +236,23 @@ func (r *Replica) syncOnce(ctx context.Context, maxSyncLTXFiles int) (result rep
return result, nil
}
if err := r.uploadLTXFile(ctx, 0, txID, txID); err != nil {
return result, err
var ltxErr *LTXError
gapMaxTXID, gapExists := findL0GapMaxTXID(r.remoteL0Files, txID)
if !errors.As(err, &ltxErr) || ltxErr.Op != "open" ||
ltxErr.Path != r.db.LTXPath(0, txID, txID) || !os.IsNotExist(ltxErr.Err) || !gapExists {
return result, err
}

r.Logger().Warn("local L0 file missing during gap heal, forcing snapshot",
"gap_min_txid", txID.String(),
"gap_max_txid", gapMaxTXID.String())
info, err := r.db.Snapshot(ctx)
if err != nil {
return result, fmt.Errorf("snapshot for missing local L0 gap %s-%s: %w", txID, gapMaxTXID, err)
}
r.SetPos(ltx.Pos{TXID: info.MaxTXID})
result.synced = true
break
}
r.SetPos(ltx.Pos{TXID: txID})
result.synced = true
Expand Down Expand Up @@ -288,10 +304,10 @@ func (r *Replica) uploadLTXFile(ctx context.Context, level int, minTXID, maxTXID
}

// calcPos returns the position replication should resume from for level 0.
// Files already compacted into level 1 are ignored. If a TXID gap exists in
// the remaining L0 files, the returned position stops just before the gap so
// the missing files are re-uploaded from disk; resuming from the maximum
// TXID would leave the gap in place permanently and block compaction.
// Files already covered by level 1 or a snapshot are ignored. If a TXID gap
// exists in the remaining L0 files, the returned position stops just before
// the gap so the missing files are re-uploaded from disk; resuming from the
// maximum TXID would leave the gap in place permanently and block compaction.
func (r *Replica) calcPos(ctx context.Context) (pos ltx.Pos, l0Files []ltx.FileInfo, err error) {
l1Info, err := r.MaxLTXFileInfo(ctx, 1)
if err != nil {
Expand All @@ -315,9 +331,17 @@ func (r *Replica) calcPos(ctx context.Context) (pos ltx.Pos, l0Files []ltx.FileI
txID := l1Info.MaxTXID
for _, info := range l0Files {
if info.MaxTXID <= txID {
continue // already compacted into L1
continue // already covered by L1 or a snapshot
}
if txID != 0 && info.MinTXID > txID+1 {
snapshotInfo, err := r.db.MaxLTXFileInfo(ctx, SnapshotLevel)
if err != nil {
return pos, nil, fmt.Errorf("max snapshot ltx file: %w", err)
}
if snapshotInfo.MaxTXID >= info.MinTXID-1 {
txID = max(snapshotInfo.MaxTXID, info.MaxTXID)
continue
}
r.Logger().Warn("txid gap detected in remote L0 files, resuming replication before gap",
"expected", (txID + 1).String(),
"actual", info.MinTXID.String())
Expand All @@ -333,6 +357,15 @@ func containsL0TXID(files []ltx.FileInfo, txID ltx.TXID) bool {
return i > 0 && files[i-1].MaxTXID >= txID
}

func findL0GapMaxTXID(files []ltx.FileInfo, minTXID ltx.TXID) (ltx.TXID, bool) {
for _, info := range files {
if info.MinTXID > minTXID {
return info.MinTXID - 1, true
}
}
return 0, false
}

// InvalidatePos marks the replica position for recalculation on the next
// sync. Used when a TXID gap is detected in remote L0 files so the next sync
// re-uploads the missing files from disk.
Expand Down
136 changes: 135 additions & 1 deletion replica_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"log/slog"
"os"
"path/filepath"
"slices"
"strings"
"testing"
"time"
Expand All @@ -24,12 +25,15 @@ import (

type l0WriteRecordingClient struct {
litestream.ReplicaClient
txIDs []ltx.TXID
txIDs []ltx.TXID
snapshotTXIDs []ltx.TXID
}

func (c *l0WriteRecordingClient) WriteLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, r io.Reader) (*ltx.FileInfo, error) {
if level == 0 {
c.txIDs = append(c.txIDs, minTXID)
} else if level == litestream.SnapshotLevel {
c.snapshotTXIDs = append(c.snapshotTXIDs, maxTXID)
}
return c.ReplicaClient.WriteLTXFile(ctx, level, minTXID, maxTXID, r)
}
Expand Down Expand Up @@ -99,6 +103,136 @@ func TestReplica_InvalidatePos_HealsL0Gap(t *testing.T) {
}
}

func TestReplica_InvalidatePos_MissingLocalL0FallsBackToSnapshot(t *testing.T) {
db, sqldb := testingutil.MustOpenDBs(t)
defer testingutil.MustCloseDBs(t, db, sqldb)

if err := db.Sync(t.Context()); err != nil {
t.Fatal(err)
}
if _, err := sqldb.ExecContext(t.Context(), `CREATE TABLE t (id INT)`); err != nil {
t.Fatal(err)
}
if err := db.Sync(t.Context()); err != nil {
t.Fatal(err)
}
for i := range 3 {
if _, err := sqldb.ExecContext(t.Context(), `INSERT INTO t (id) VALUES (?)`, i); err != nil {
t.Fatal(err)
}
if err := db.Sync(t.Context()); err != nil {
t.Fatal(err)
}
}
if err := db.Replica.Sync(t.Context()); err != nil {
t.Fatal(err)
}

dpos, err := db.Pos()
if err != nil {
t.Fatal(err)
}
if dpos.TXID < 5 {
t.Fatalf("expected at least 5 transactions, got %s", dpos.TXID)
}

gapMinTXID, gapMaxTXID := dpos.TXID-2, dpos.TXID-1
if err := db.Replica.Client.DeleteLTXFiles(t.Context(), []*ltx.FileInfo{
{Level: 0, MinTXID: gapMinTXID, MaxTXID: gapMinTXID},
{Level: 0, MinTXID: gapMaxTXID, MaxTXID: gapMaxTXID},
}); err != nil {
t.Fatal(err)
}
for txID := gapMinTXID; txID <= gapMaxTXID; txID++ {
if err := os.Remove(db.LTXPath(0, txID, txID)); err != nil {
t.Fatal(err)
}
}

client := &l0WriteRecordingClient{ReplicaClient: db.Replica.Client}
db.Replica.Client = client
var logBuf bytes.Buffer
db.SetLogger(slog.New(slog.NewTextHandler(&logBuf, nil)))

db.Replica.InvalidatePos()
if err := db.Replica.Sync(t.Context()); err != nil {
t.Fatal(err)
}
if got, want := db.Replica.Pos().TXID, dpos.TXID; got != want {
t.Fatalf("replica pos=%s, want %s", got, want)
}
if got, want := client.snapshotTXIDs, []ltx.TXID{dpos.TXID}; !slices.Equal(got, want) {
t.Fatalf("snapshot TXIDs=%v, want %v", got, want)
}
for _, field := range []string{
`level=WARN`,
`msg="local L0 file missing during gap heal, forcing snapshot"`,
`gap_min_txid=` + gapMinTXID.String(),
`gap_max_txid=` + gapMaxTXID.String(),
} {
if !strings.Contains(logBuf.String(), field) {
t.Fatalf("fallback log missing %q: %s", field, logBuf.String())
}
}

db.Replica.InvalidatePos()
if err := db.Replica.Sync(t.Context()); err != nil {
t.Fatal(err)
}
if got, want := client.snapshotTXIDs, []ltx.TXID{dpos.TXID}; !slices.Equal(got, want) {
t.Fatalf("snapshot TXIDs after recalculation=%v, want %v", got, want)
}

plan, err := litestream.CalcRestorePlan(t.Context(), client, 0, time.Time{}, db.Logger)
if err != nil {
t.Fatal(err)
}
if got, want := len(plan), 1; got != want {
t.Fatalf("restore plan length=%d, want %d: %#v", got, want, plan)
}
if got, want := plan[0].Level, litestream.SnapshotLevel; got != want {
t.Fatalf("restore plan level=%d, want %d", got, want)
}
if got, want := plan[0].MinTXID, ltx.TXID(1); got != want {
t.Fatalf("restore plan min TXID=%s, want %s", got, want)
}
if got, want := plan[0].MaxTXID, dpos.TXID; got != want {
t.Fatalf("restore plan max TXID=%s, want %s", got, want)
}

restorePath := filepath.Join(t.TempDir(), "restore.db")
opt := litestream.NewRestoreOptions()
opt.OutputPath = restorePath
if err := db.Replica.Restore(t.Context(), opt); err != nil {
t.Fatal(err)
}
restoredDB := testingutil.MustOpenSQLDB(t, restorePath)
defer testingutil.MustCloseSQLDB(t, restoredDB)
var rowN, idSum int
if err := restoredDB.QueryRowContext(t.Context(), `SELECT COUNT(*), SUM(id) FROM t`).Scan(&rowN, &idSum); err != nil {
t.Fatal(err)
}
if got, want := rowN, 3; got != want {
t.Fatalf("restored row count=%d, want %d", got, want)
}
if got, want := idSum, 3; got != want {
t.Fatalf("restored id sum=%d, want %d", got, want)
}

if _, err := sqldb.ExecContext(t.Context(), `INSERT INTO t (id) VALUES (3)`); err != nil {
t.Fatal(err)
}
if err := db.Sync(t.Context()); err != nil {
t.Fatal(err)
}
if err := db.Replica.Sync(t.Context()); err != nil {
t.Fatal(err)
}
if got, want := db.Replica.Pos().TXID, dpos.TXID+1; got != want {
t.Fatalf("replica pos after new write=%s, want %s", got, want)
}
}

func TestReplica_Sync(t *testing.T) {
db, sqldb := testingutil.MustOpenDBs(t)
defer testingutil.MustCloseDBs(t, db, sqldb)
Expand Down
Loading