diff --git a/db.go b/db.go index 0c2f3a8b3..6dce068d0 100644 --- a/db.go +++ b/db.go @@ -707,7 +707,7 @@ func (db *DB) SyncStatus(ctx context.Context) (SyncStatus, error) { return SyncStatus{}, fmt.Errorf("local position: %w", err) } - remotePos, err := db.Replica.calcPos(ctx) + remotePos, _, err := db.Replica.calcPos(ctx) if err != nil { return SyncStatus{}, fmt.Errorf("remote position: %w", err) } diff --git a/replica.go b/replica.go index 8b13f3573..4791cb1b6 100644 --- a/replica.go +++ b/replica.go @@ -36,8 +36,9 @@ var errReplicaWaitForData = errors.New("no position, waiting for data") type Replica struct { db *DB - mu sync.RWMutex - pos ltx.Pos // current replicated position + mu sync.RWMutex + pos ltx.Pos // current replicated position + remoteL0Files []ltx.FileInfo syncSem *semaphore.Weighted syncWaiters atomic.Int64 // diagnostic instrumentation: goroutines queued on syncSem @@ -175,6 +176,7 @@ func (r *Replica) syncOnce(ctx context.Context, maxSyncLTXFiles int) (result rep r.mu.Lock() r.pos = ltx.Pos{} r.mu.Unlock() + r.remoteL0Files = nil } }() @@ -186,15 +188,17 @@ func (r *Replica) syncOnce(ctx context.Context, maxSyncLTXFiles int) (result rep // such as when compaction detects a TXID gap in remote L0 files. if r.posInvalid.Swap(false) { r.SetPos(ltx.Pos{}) + r.remoteL0Files = nil } // Calculate current replica position, if unknown. if r.Pos().IsZero() { - pos, err := r.calcPos(ctx) + pos, remoteL0Files, err := r.calcPos(ctx) if err != nil { return result, fmt.Errorf("calc pos: %w", err) } r.SetPos(pos) + r.remoteL0Files = remoteL0Files } // Find current position of database. @@ -213,6 +217,13 @@ func (r *Replica) syncOnce(ctx context.Context, maxSyncLTXFiles int) (result rep // Replicate all L0 LTX files since last replica position. for txID, syncedFileN := r.Pos().TXID+1, 0; txID <= dpos.TXID; txID = r.Pos().TXID + 1 { + if err := ctx.Err(); err != nil { + return result, context.Cause(ctx) + } + if containsL0TXID(r.remoteL0Files, txID) { + r.SetPos(ltx.Pos{TXID: txID}) + continue + } if maxSyncLTXFiles > 0 && syncedFileN >= maxSyncLTXFiles { result.limited = true // Uploads succeeded, so record sync health; otherwise a @@ -224,9 +235,6 @@ func (r *Replica) syncOnce(ctx context.Context, maxSyncLTXFiles int) (result rep "replica_txid", r.Pos().TXID.String()) return result, nil } - if err := ctx.Err(); err != nil { - return result, context.Cause(ctx) - } if err := r.uploadLTXFile(ctx, 0, txID, txID); err != nil { return result, err } @@ -234,6 +242,7 @@ func (r *Replica) syncOnce(ctx context.Context, maxSyncLTXFiles int) (result rep result.synced = true syncedFileN++ } + r.remoteL0Files = nil // Record successful sync for heartbeat monitoring. r.db.RecordSuccessfulSync() @@ -283,29 +292,28 @@ func (r *Replica) uploadLTXFile(ctx context.Context, level int, minTXID, maxTXID // 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, err error) { +func (r *Replica) calcPos(ctx context.Context) (pos ltx.Pos, l0Files []ltx.FileInfo, err error) { l1Info, err := r.MaxLTXFileInfo(ctx, 1) if err != nil { - return pos, fmt.Errorf("max l1 ltx file: %w", err) + return pos, nil, fmt.Errorf("max l1 ltx file: %w", err) } itr, err := r.Client.LTXFiles(ctx, 0, 0, false) if err != nil { - return pos, fmt.Errorf("l0 ltx files: %w", err) + return pos, nil, fmt.Errorf("l0 ltx files: %w", err) } defer itr.Close() - var infos []ltx.FileInfo for itr.Next() { - infos = append(infos, *itr.Item()) + l0Files = append(l0Files, *itr.Item()) } if err := itr.Close(); err != nil { - return pos, fmt.Errorf("close l0 iterator: %w", err) + return pos, nil, fmt.Errorf("close l0 iterator: %w", err) } - sort.Slice(infos, func(i, j int) bool { return infos[i].MinTXID < infos[j].MinTXID }) + sort.Slice(l0Files, func(i, j int) bool { return l0Files[i].MinTXID < l0Files[j].MinTXID }) txID := l1Info.MaxTXID - for _, info := range infos { + for _, info := range l0Files { if info.MaxTXID <= txID { continue // already compacted into L1 } @@ -317,7 +325,12 @@ func (r *Replica) calcPos(ctx context.Context) (pos ltx.Pos, err error) { } txID = info.MaxTXID } - return ltx.Pos{TXID: txID}, nil + return ltx.Pos{TXID: txID}, l0Files, nil +} + +func containsL0TXID(files []ltx.FileInfo, txID ltx.TXID) bool { + i := sort.Search(len(files), func(i int) bool { return files[i].MinTXID > txID }) + return i > 0 && files[i-1].MaxTXID >= txID } // InvalidatePos marks the replica position for recalculation on the next diff --git a/replica_test.go b/replica_test.go index 89b8b0a07..169112af6 100644 --- a/replica_test.go +++ b/replica_test.go @@ -22,6 +22,18 @@ import ( "github.com/benbjohnson/litestream/mock" ) +type l0WriteRecordingClient struct { + litestream.ReplicaClient + txIDs []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) + } + return c.ReplicaClient.WriteLTXFile(ctx, level, minTXID, maxTXID, r) +} + func TestReplica_InvalidatePos_HealsL0Gap(t *testing.T) { db, sqldb := testingutil.MustOpenDBs(t) defer testingutil.MustCloseDBs(t, db, sqldb) @@ -63,6 +75,8 @@ func TestReplica_InvalidatePos_HealsL0Gap(t *testing.T) { t.Fatal(err) } + client := &l0WriteRecordingClient{ReplicaClient: db.Replica.Client} + db.Replica.Client = client db.Replica.InvalidatePos() if err := db.Replica.Sync(t.Context()); err != nil { t.Fatal(err) @@ -77,6 +91,12 @@ func TestReplica_InvalidatePos_HealsL0Gap(t *testing.T) { if got, want := db.Replica.Pos().TXID, dpos.TXID; got != want { t.Fatalf("replica pos=%s, want %s", got, want) } + if got, want := len(client.txIDs), 1; got != want { + t.Fatalf("L0 write count=%d, want %d; txids=%v", got, want, client.txIDs) + } + if got, want := client.txIDs[0], gapTXID; got != want { + t.Fatalf("L0 write TXID=%s, want %s", got, want) + } } func TestReplica_Sync(t *testing.T) {