Reduce snapshot check frequency

Previously, a bug was introduced that added a `LIST` operation
on every replica sync which significantly increased the cost of
running Litestream against S3. This changes the behavior to only
issue the `LIST` operation when the generation has changed.

abs/replica_client.go

@@ -102,7 +102,7 @@ func (c *ReplicaClient) Generations(ctx context.Context) ([]string, error) {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
 		resp, err := c.containerURL.ListBlobsHierarchySegment(ctx, marker, "/", azblob.ListBlobsSegmentOptions{
-			Prefix: path.Join(c.Path, "generations") + "/",
+			Prefix: litestream.GenerationsPath(c.Path) + "/",
 		})
 		if err != nil {
 			return nil, err
@@ -125,17 +125,18 @@ func (c *ReplicaClient) Generations(ctx context.Context) ([]string, error) {
 func (c *ReplicaClient) DeleteGeneration(ctx context.Context, generation string) error {
 	if err := c.Init(ctx); err != nil {
 		return err
-	} else if generation == "" {
-		return fmt.Errorf("generation required")
 	}
 
-	prefix := path.Join(c.Path, "generations", generation) + "/"
+	dir, err := litestream.GenerationPath(c.Path, generation)
+	if err != nil {
+		return fmt.Errorf("cannot determine generation path: %w", err)
+	}
 
 	var marker azblob.Marker
 	for marker.NotDone() {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
-		resp, err := c.containerURL.ListBlobsFlatSegment(ctx, marker, azblob.ListBlobsSegmentOptions{Prefix: prefix})
+		resp, err := c.containerURL.ListBlobsFlatSegment(ctx, marker, azblob.ListBlobsSegmentOptions{Prefix: dir + "/"})
 		if err != nil {
 			return err
 		}
@@ -170,11 +171,12 @@ func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (lites
 func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, index int, rd io.Reader) (info litestream.SnapshotInfo, err error) {
 	if err := c.Init(ctx); err != nil {
 		return info, err
-	} else if generation == "" {
-		return info, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	key, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return info, fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 	startTime := time.Now()
 
 	rc := internal.NewReadCounter(rd)
@@ -204,11 +206,12 @@ func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, in
 func (c *ReplicaClient) SnapshotReader(ctx context.Context, generation string, index int) (io.ReadCloser, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
-	} else if generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	key, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 
 	blobURL := c.containerURL.NewBlobURL(key)
 	resp, err := blobURL.Download(ctx, 0, 0, azblob.BlobAccessConditions{}, false, azblob.ClientProvidedKeyOptions{})
@@ -228,11 +231,12 @@ func (c *ReplicaClient) SnapshotReader(ctx context.Context, generation string, i
 func (c *ReplicaClient) DeleteSnapshot(ctx context.Context, generation string, index int) error {
 	if err := c.Init(ctx); err != nil {
 		return err
-	} else if generation == "" {
-		return fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	key, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 
 	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "DELETE").Inc()
 
@@ -257,11 +261,12 @@ func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (lit
 func (c *ReplicaClient) WriteWALSegment(ctx context.Context, pos litestream.Pos, rd io.Reader) (info litestream.WALSegmentInfo, err error) {
 	if err := c.Init(ctx); err != nil {
 		return info, err
-	} else if pos.Generation == "" {
-		return info, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
+	key, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+	if err != nil {
+		return info, fmt.Errorf("cannot determine wal segment path: %w", err)
+	}
 	startTime := time.Now()
 
 	rc := internal.NewReadCounter(rd)
@@ -291,11 +296,12 @@ func (c *ReplicaClient) WriteWALSegment(ctx context.Context, pos litestream.Pos,
 func (c *ReplicaClient) WALSegmentReader(ctx context.Context, pos litestream.Pos) (io.ReadCloser, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
-	} else if pos.Generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
+	key, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine wal segment path: %w", err)
+	}
 
 	blobURL := c.containerURL.NewBlobURL(key)
 	resp, err := blobURL.Download(ctx, 0, 0, azblob.BlobAccessConditions{}, false, azblob.ClientProvidedKeyOptions{})
@@ -318,12 +324,11 @@ func (c *ReplicaClient) DeleteWALSegments(ctx context.Context, a []litestream.Po
 	}
 
 	for _, pos := range a {
-		if pos.Generation == "" {
-			return fmt.Errorf("generation required")
+		key, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+		if err != nil {
+			return fmt.Errorf("cannot determine wal segment path: %w", err)
 		}
 
-		key := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
-
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "DELETE").Inc()
 
 		blobURL := c.containerURL.NewBlobURL(key)
@@ -367,24 +372,24 @@ func newSnapshotIterator(ctx context.Context, generation string, client *Replica
 func (itr *snapshotIterator) fetch() error {
 	defer close(itr.ch)
 
-	if itr.generation == "" {
-		return fmt.Errorf("generation required")
+	dir, err := litestream.SnapshotsPath(itr.client.Path, itr.generation)
+	if err != nil {
+		return fmt.Errorf("cannot determine snapshots path: %w", err)
 	}
 
-	prefix := path.Join(itr.client.Path, "generations", itr.generation) + "/"
-
 	var marker azblob.Marker
 	for marker.NotDone() {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
-		resp, err := itr.client.containerURL.ListBlobsFlatSegment(itr.ctx, marker, azblob.ListBlobsSegmentOptions{Prefix: prefix})
+		resp, err := itr.client.containerURL.ListBlobsFlatSegment(itr.ctx, marker, azblob.ListBlobsSegmentOptions{Prefix: dir + "/"})
 		if err != nil {
 			return err
 		}
 		marker = resp.NextMarker
 
 		for _, item := range resp.Segment.BlobItems {
-			index, err := internal.ParseSnapshotPath(path.Base(item.Name))
+			key := path.Base(item.Name)
+			index, err := litestream.ParseSnapshotPath(key)
 			if err != nil {
 				continue
 			}
@@ -473,24 +478,24 @@ func newWALSegmentIterator(ctx context.Context, generation string, client *Repli
 func (itr *walSegmentIterator) fetch() error {
 	defer close(itr.ch)
 
-	if itr.generation == "" {
-		return fmt.Errorf("generation required")
+	dir, err := litestream.WALPath(itr.client.Path, itr.generation)
+	if err != nil {
+		return fmt.Errorf("cannot determine wal path: %w", err)
 	}
-	prefix := path.Join(itr.client.Path, "generations", itr.generation, "wal")
 
 	var marker azblob.Marker
 	for marker.NotDone() {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
-		resp, err := itr.client.containerURL.ListBlobsFlatSegment(itr.ctx, marker, azblob.ListBlobsSegmentOptions{Prefix: prefix})
+		resp, err := itr.client.containerURL.ListBlobsFlatSegment(itr.ctx, marker, azblob.ListBlobsSegmentOptions{Prefix: dir + "/"})
 		if err != nil {
 			return err
 		}
 		marker = resp.NextMarker
 
 		for _, item := range resp.Segment.BlobItems {
-			key := strings.TrimPrefix(item.Name, prefix+"/")
-			index, offset, err := internal.ParseWALSegmentPath(key)
+			key := path.Base(item.Name)
+			index, offset, err := litestream.ParseWALSegmentPath(key)
 			if err != nil {
 				continue
 			}

db.go

@@ -16,8 +16,6 @@ import (
 	"math/rand"
 	"os"
 	"path/filepath"
-	"regexp"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -170,7 +168,7 @@ func (db *DB) ShadowWALDir(generation string) string {
 // Panics if generation is blank or index is negative.
 func (db *DB) ShadowWALPath(generation string, index int) string {
 	assert(index >= 0, "shadow wal index cannot be negative")
-	return filepath.Join(db.ShadowWALDir(generation), FormatIndex(index)+".wal")
+	return filepath.Join(db.ShadowWALDir(generation), FormatWALPath(index))
 }
 
 // CurrentShadowWALPath returns the path to the last shadow WAL in a generation.
@@ -193,8 +191,8 @@ func (db *DB) CurrentShadowWALIndex(generation string) (index int, size int64, e
 
 	// Find highest wal index.
 	for _, fi := range fis {
-		if v, err := parseWALPath(fi.Name()); err != nil {
-			continue // invalid filename
+		if v, err := ParseWALPath(fi.Name()); err != nil {
+			continue // invalid wal filename
 		} else if v > index {
 			index = v
 		}
@@ -586,7 +584,7 @@ func (db *DB) cleanWAL() error {
 		return err
 	}
 	for _, fi := range fis {
-		if idx, err := parseWALPath(fi.Name()); err != nil || idx >= min {
+		if idx, err := ParseWALPath(fi.Name()); err != nil || idx >= min {
 			continue
 		}
 		if err := os.Remove(filepath.Join(dir, fi.Name())); err != nil {
@@ -930,13 +928,13 @@ func (db *DB) syncWAL(info syncInfo) (newSize int64, err error) {
 
 	// Parse index of current shadow WAL file.
 	dir, base := filepath.Split(info.shadowWALPath)
-	index, err := parseWALPath(base)
+	index, err := ParseWALPath(base)
 	if err != nil {
 		return 0, fmt.Errorf("cannot parse shadow wal filename: %s", base)
 	}
 
 	// Start a new shadow WAL file with next index.
-	newShadowWALPath := filepath.Join(dir, formatWALPath(index+1))
+	newShadowWALPath := filepath.Join(dir, FormatWALPath(index+1))
 	newSize, err = db.initShadowWALFile(newShadowWALPath)
 	if err != nil {
 		return 0, fmt.Errorf("cannot init shadow wal file: name=%s err=%w", newShadowWALPath, err)
@@ -1300,13 +1298,13 @@ func (db *DB) checkpoint(ctx context.Context, generation, mode string) error {
 	}
 
 	// Parse index of current shadow WAL file.
-	index, err := parseWALPath(shadowWALPath)
+	index, err := ParseWALPath(shadowWALPath)
 	if err != nil {
 		return fmt.Errorf("cannot parse shadow wal filename: %s", shadowWALPath)
 	}
 
 	// Start a new shadow WAL file with next index.
-	newShadowWALPath := filepath.Join(filepath.Dir(shadowWALPath), formatWALPath(index+1))
+	newShadowWALPath := filepath.Join(filepath.Dir(shadowWALPath), FormatWALPath(index+1))
 	if _, err := db.initShadowWALFile(newShadowWALPath); err != nil {
 		return fmt.Errorf("cannot init shadow wal file: name=%s err=%w", newShadowWALPath, err)
 	}
@@ -1483,28 +1481,6 @@ func (db *DB) CRC64(ctx context.Context) (uint64, Pos, error) {
 	return h.Sum64(), pos, nil
 }
 
-// parseWALPath returns the index for the WAL file.
-// Returns an error if the path is not a valid WAL path.
-func parseWALPath(s string) (index int, err error) {
-	s = filepath.Base(s)
-
-	a := walPathRegex.FindStringSubmatch(s)
-	if a == nil {
-		return 0, fmt.Errorf("invalid wal path: %s", s)
-	}
-
-	i64, _ := strconv.ParseUint(a[1], 16, 64)
-	return int(i64), nil
-}
-
-// formatWALPath formats a WAL filename with a given index.
-func formatWALPath(index int) string {
-	assert(index >= 0, "wal index must be non-negative")
-	return FormatIndex(index) + ".wal"
-}
-
-var walPathRegex = regexp.MustCompile(`^([0-9a-f]{8})\.wal$`)
-
 // DefaultRestoreParallelism is the default parallelism when downloading WAL files.
 const DefaultRestoreParallelism = 8
 

file/replica_client.go

@@ -8,7 +8,6 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
-	"strings"
 
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/internal"
@@ -85,7 +84,7 @@ func (c *ReplicaClient) SnapshotPath(generation string, index int) (string, erro
 	if err != nil {
 		return "", err
 	}
-	return filepath.Join(dir, litestream.FormatIndex(index)+".snapshot.lz4"), nil
+	return filepath.Join(dir, litestream.FormatSnapshotPath(index)), nil
 }
 
 // WALDir returns the path to a generation's WAL directory
@@ -103,7 +102,7 @@ func (c *ReplicaClient) WALSegmentPath(generation string, index int, offset int6
 	if err != nil {
 		return "", err
 	}
-	return filepath.Join(dir, litestream.FormatIndex(index), fmt.Sprintf("%08x.wal.lz4", offset)), nil
+	return filepath.Join(dir, litestream.FormatWALSegmentPath(index, offset)), nil
 }
 
 // Generations returns a list of available generation names.
@@ -169,7 +168,7 @@ func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (lites
 	infos := make([]litestream.SnapshotInfo, 0, len(fis))
 	for _, fi := range fis {
 		// Parse index from filename.
-		index, err := internal.ParseSnapshotPath(filepath.Base(fi.Name()))
+		index, err := litestream.ParseSnapshotPath(fi.Name())
 		if err != nil {
 			continue
 		}
@@ -282,18 +281,26 @@ func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (lit
 	}
 
 	// Iterate over every file and convert to metadata.
-	indexes := make([]int, 0, len(fis))
+	infos := make([]litestream.WALSegmentInfo, 0, len(fis))
 	for _, fi := range fis {
-		index, err := litestream.ParseIndex(fi.Name())
-		if err != nil || !fi.IsDir() {
+		// Parse index from filename.
+		index, offset, err := litestream.ParseWALSegmentPath(fi.Name())
+		if err != nil {
 			continue
 		}
-		indexes = append(indexes, index)
+
+		infos = append(infos, litestream.WALSegmentInfo{
+			Generation: generation,
+			Index:      index,
+			Offset:     offset,
+			Size:       fi.Size(),
+			CreatedAt:  fi.ModTime().UTC(),
+		})
 	}
 
-	sort.Ints(indexes)
+	sort.Sort(litestream.WALSegmentInfoSlice(infos))
 
-	return newWALSegmentIterator(dir, generation, indexes), nil
+	return litestream.NewWALSegmentInfoSliceIterator(infos), nil
 }
 
 // WriteWALSegment writes LZ4 compressed data from rd into a file on disk.
@@ -372,97 +379,3 @@ func (c *ReplicaClient) DeleteWALSegments(ctx context.Context, a []litestream.Po
 	}
 	return nil
 }
-
-type walSegmentIterator struct {
-	dir        string
-	generation string
-	indexes    []int
-
-	infos []litestream.WALSegmentInfo
-	err   error
-}
-
-func newWALSegmentIterator(dir, generation string, indexes []int) *walSegmentIterator {
-	return &walSegmentIterator{
-		dir:        dir,
-		generation: generation,
-		indexes:    indexes,
-	}
-}
-
-func (itr *walSegmentIterator) Close() (err error) {
-	return itr.err
-}
-
-func (itr *walSegmentIterator) Next() bool {
-	// Exit if an error has already occurred.
-	if itr.err != nil {
-		return false
-	}
-
-	for {
-		// Move to the next segment in cache, if available.
-		if len(itr.infos) > 1 {
-			itr.infos = itr.infos[1:]
-			return true
-		}
-		itr.infos = itr.infos[:0] // otherwise clear infos
-
-		// Move to the next index unless this is the first time initializing.
-		if itr.infos != nil && len(itr.indexes) > 0 {
-			itr.indexes = itr.indexes[1:]
-		}
-
-		// If no indexes remain, stop iteration.
-		if len(itr.indexes) == 0 {
-			return false
-		}
-
-		// Read segments into a cache for the current index.
-		index := itr.indexes[0]
-		f, err := os.Open(filepath.Join(itr.dir, litestream.FormatIndex(index)))
-		if err != nil {
-			itr.err = err
-			return false
-		}
-		defer f.Close()
-
-		fis, err := f.Readdir(-1)
-		if err != nil {
-			itr.err = err
-			return false
-		}
-		for _, fi := range fis {
-			filename := filepath.Base(fi.Name())
-			if fi.IsDir() {
-				continue
-			}
-
-			offset, err := litestream.ParseOffset(strings.TrimSuffix(filename, ".wal.lz4"))
-			if err != nil {
-				continue
-			}
-
-			itr.infos = append(itr.infos, litestream.WALSegmentInfo{
-				Generation: itr.generation,
-				Index:      index,
-				Offset:     offset,
-				Size:       fi.Size(),
-				CreatedAt:  fi.ModTime().UTC(),
-			})
-		}
-
-		if len(itr.infos) > 0 {
-			return true
-		}
-	}
-}
-
-func (itr *walSegmentIterator) Err() error { return itr.err }
-
-func (itr *walSegmentIterator) WALSegment() litestream.WALSegmentInfo {
-	if len(itr.infos) == 0 {
-		return litestream.WALSegmentInfo{}
-	}
-	return itr.infos[0]
-}

file/replica_client_test.go

@@ -118,7 +118,7 @@ func TestReplicaClient_WALSegmentPath(t *testing.T) {
 	t.Run("OK", func(t *testing.T) {
 		if got, err := file.NewReplicaClient("/foo").WALSegmentPath("0123456701234567", 1000, 1001); err != nil {
 			t.Fatal(err)
-		} else if want := "/foo/generations/0123456701234567/wal/000003e8/000003e9.wal.lz4"; got != want {
+		} else if want := "/foo/generations/0123456701234567/wal/000003e8_000003e9.wal.lz4"; got != want {
 			t.Fatalf("WALPath()=%v, want %v", got, want)
 		}
 	})

gcs/replica_client.go

@@ -68,7 +68,7 @@ func (c *ReplicaClient) Generations(ctx context.Context) ([]string, error) {
 	// Construct query to only pull generation directory names.
 	query := &storage.Query{
 		Delimiter: "/",
-		Prefix:    path.Join(c.Path, "generations") + "/",
+		Prefix:    litestream.GenerationsPath(c.Path) + "/",
 	}
 
 	// Loop over results and only build list of generation-formatted names.
@@ -96,15 +96,16 @@ func (c *ReplicaClient) Generations(ctx context.Context) ([]string, error) {
 func (c *ReplicaClient) DeleteGeneration(ctx context.Context, generation string) error {
 	if err := c.Init(ctx); err != nil {
 		return err
-	} else if generation == "" {
-		return fmt.Errorf("generation required")
 	}
 
-	prefix := path.Join(c.Path, "generations", generation) + "/"
+	dir, err := litestream.GenerationPath(c.Path, generation)
+	if err != nil {
+		return fmt.Errorf("cannot determine generation path: %w", err)
+	}
 
 	// Iterate over every object in generation and delete it.
 	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
-	for it := c.bkt.Objects(ctx, &storage.Query{Prefix: prefix}); ; {
+	for it := c.bkt.Objects(ctx, &storage.Query{Prefix: dir + "/"}); ; {
 		attrs, err := it.Next()
 		if err == iterator.Done {
 			break
@@ -129,22 +130,24 @@ func (c *ReplicaClient) DeleteGeneration(ctx context.Context, generation string)
 func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (litestream.SnapshotIterator, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
-	} else if generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
-	prefix := path.Join(c.Path, "generations", generation) + "/"
-	return newSnapshotIterator(generation, c.bkt.Objects(ctx, &storage.Query{Prefix: prefix})), nil
+	dir, err := litestream.SnapshotsPath(c.Path, generation)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine snapshots path: %w", err)
+	}
+	return newSnapshotIterator(generation, c.bkt.Objects(ctx, &storage.Query{Prefix: dir + "/"})), nil
 }
 
 // WriteSnapshot writes LZ4 compressed data from rd to the object storage.
 func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, index int, rd io.Reader) (info litestream.SnapshotInfo, err error) {
 	if err := c.Init(ctx); err != nil {
 		return info, err
-	} else if generation == "" {
-		return info, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	key, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return info, fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 	startTime := time.Now()
 
 	w := c.bkt.Object(key).NewWriter(ctx)
@@ -174,11 +177,12 @@ func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, in
 func (c *ReplicaClient) SnapshotReader(ctx context.Context, generation string, index int) (io.ReadCloser, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
-	} else if generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	key, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 
 	r, err := c.bkt.Object(key).NewReader(ctx)
 	if isNotExists(err) {
@@ -197,11 +201,12 @@ func (c *ReplicaClient) SnapshotReader(ctx context.Context, generation string, i
 func (c *ReplicaClient) DeleteSnapshot(ctx context.Context, generation string, index int) error {
 	if err := c.Init(ctx); err != nil {
 		return err
-	} else if generation == "" {
-		return fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index), ".snapshot.lz4")
+	key, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 
 	if err := c.bkt.Object(key).Delete(ctx); err != nil && !isNotExists(err) {
 		return fmt.Errorf("cannot delete snapshot %q: %w", key, err)
@@ -215,22 +220,24 @@ func (c *ReplicaClient) DeleteSnapshot(ctx context.Context, generation string, i
 func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (litestream.WALSegmentIterator, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
-	} else if generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
-	prefix := path.Join(c.Path, "generations", generation, "wal") + "/"
-	return newWALSegmentIterator(generation, prefix, c.bkt.Objects(ctx, &storage.Query{Prefix: prefix})), nil
+	dir, err := litestream.WALPath(c.Path, generation)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine wal path: %w", err)
+	}
+	return newWALSegmentIterator(generation, c.bkt.Objects(ctx, &storage.Query{Prefix: dir + "/"})), nil
 }
 
 // WriteWALSegment writes LZ4 compressed data from rd into a file on disk.
 func (c *ReplicaClient) WriteWALSegment(ctx context.Context, pos litestream.Pos, rd io.Reader) (info litestream.WALSegmentInfo, err error) {
 	if err := c.Init(ctx); err != nil {
 		return info, err
-	} else if pos.Generation == "" {
-		return info, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
+	key, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+	if err != nil {
+		return info, fmt.Errorf("cannot determine wal segment path: %w", err)
+	}
 	startTime := time.Now()
 
 	w := c.bkt.Object(key).NewWriter(ctx)
@@ -260,11 +267,12 @@ func (c *ReplicaClient) WriteWALSegment(ctx context.Context, pos litestream.Pos,
 func (c *ReplicaClient) WALSegmentReader(ctx context.Context, pos litestream.Pos) (io.ReadCloser, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
-	} else if pos.Generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
+	key, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine wal segment path: %w", err)
+	}
 
 	r, err := c.bkt.Object(key).NewReader(ctx)
 	if isNotExists(err) {
@@ -286,11 +294,11 @@ func (c *ReplicaClient) DeleteWALSegments(ctx context.Context, a []litestream.Po
 	}
 
 	for _, pos := range a {
-		if pos.Generation == "" {
-			return fmt.Errorf("generation required")
+		key, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+		if err != nil {
+			return fmt.Errorf("cannot determine wal segment path: %w", err)
 		}
 
-		key := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
 		if err := c.bkt.Object(key).Delete(ctx); err != nil && !isNotExists(err) {
 			return fmt.Errorf("cannot delete wal segment %q: %w", key, err)
 		}
@@ -336,7 +344,7 @@ func (itr *snapshotIterator) Next() bool {
 		}
 
 		// Parse index, otherwise skip to the next object.
-		index, err := internal.ParseSnapshotPath(path.Base(attrs.Name))
+		index, err := litestream.ParseSnapshotPath(path.Base(attrs.Name))
 		if err != nil {
 			continue
 		}
@@ -358,17 +366,15 @@ func (itr *snapshotIterator) Snapshot() litestream.SnapshotInfo { return itr.inf
 
 type walSegmentIterator struct {
 	generation string
-	prefix     string
 
 	it   *storage.ObjectIterator
 	info litestream.WALSegmentInfo
 	err  error
 }
 
-func newWALSegmentIterator(generation, prefix string, it *storage.ObjectIterator) *walSegmentIterator {
+func newWALSegmentIterator(generation string, it *storage.ObjectIterator) *walSegmentIterator {
 	return &walSegmentIterator{
 		generation: generation,
-		prefix:     prefix,
 		it:         it,
 	}
 }
@@ -394,7 +400,7 @@ func (itr *walSegmentIterator) Next() bool {
 		}
 
 		// Parse index & offset, otherwise skip to the next object.
-		index, offset, err := internal.ParseWALSegmentPath(strings.TrimPrefix(attrs.Name, itr.prefix))
+		index, offset, err := litestream.ParseWALSegmentPath(path.Base(attrs.Name))
 		if err != nil {
 			continue
 		}

internal/internal.go

@@ -1,11 +1,8 @@
 package internal
 
 import (
-	"fmt"
 	"io"
 	"os"
-	"regexp"
-	"strconv"
 	"syscall"
 
 	"github.com/prometheus/client_golang/prometheus"
@@ -130,33 +127,6 @@ func MkdirAll(path string, fi os.FileInfo) error {
 	return nil
 }
 
-// ParseSnapshotPath parses the index from a snapshot filename. Used by path-based replicas.
-func ParseSnapshotPath(s string) (index int, err error) {
-	a := snapshotPathRegex.FindStringSubmatch(s)
-	if a == nil {
-		return 0, fmt.Errorf("invalid snapshot path")
-	}
-
-	i64, _ := strconv.ParseUint(a[1], 16, 64)
-	return int(i64), nil
-}
-
-var snapshotPathRegex = regexp.MustCompile(`^([0-9a-f]{8})\.snapshot\.lz4$`)
-
-// ParseWALSegmentPath parses the index/offset from a segment filename. Used by path-based replicas.
-func ParseWALSegmentPath(s string) (index int, offset int64, err error) {
-	a := walSegmentPathRegex.FindStringSubmatch(s)
-	if a == nil {
-		return 0, 0, fmt.Errorf("invalid wal segment path")
-	}
-
-	i64, _ := strconv.ParseUint(a[1], 16, 64)
-	off64, _ := strconv.ParseUint(a[2], 16, 64)
-	return int(i64), int64(off64), nil
-}
-
-var walSegmentPathRegex = regexp.MustCompile(`^([0-9a-f]{8})\/([0-9a-f]{8})\.wal\.lz4$`)
-
 // Shared replica metrics.
 var (
 	OperationTotalCounterVec = promauto.NewCounterVec(prometheus.CounterOpts{

internal/internal_test.go

@@ -1,61 +0,0 @@
-package internal_test
-
-import (
-	"fmt"
-	"reflect"
-	"testing"
-
-	"github.com/benbjohnson/litestream/internal"
-)
-
-func TestParseSnapshotPath(t *testing.T) {
-	for _, tt := range []struct {
-		s     string
-		index int
-		err   error
-	}{
-		{"00bc614e.snapshot.lz4", 12345678, nil},
-		{"xxxxxxxx.snapshot.lz4", 0, fmt.Errorf("invalid snapshot path")},
-		{"00bc614.snapshot.lz4", 0, fmt.Errorf("invalid snapshot path")},
-		{"00bc614e.snapshot.lz", 0, fmt.Errorf("invalid snapshot path")},
-		{"00bc614e.snapshot", 0, fmt.Errorf("invalid snapshot path")},
-		{"00bc614e", 0, fmt.Errorf("invalid snapshot path")},
-		{"", 0, fmt.Errorf("invalid snapshot path")},
-	} {
-		t.Run("", func(t *testing.T) {
-			index, err := internal.ParseSnapshotPath(tt.s)
-			if got, want := index, tt.index; got != want {
-				t.Errorf("index=%#v, want %#v", got, want)
-			} else if got, want := err, tt.err; !reflect.DeepEqual(got, want) {
-				t.Errorf("err=%#v, want %#v", got, want)
-			}
-		})
-	}
-}
-
-func TestParseWALSegmentPath(t *testing.T) {
-	for _, tt := range []struct {
-		s      string
-		index  int
-		offset int64
-		err    error
-	}{
-		{"00bc614e/000003e8.wal.lz4", 12345678, 1000, nil},
-		{"00000000/00000000.wal", 0, 0, fmt.Errorf("invalid wal segment path")},
-		{"00000000/00000000", 0, 0, fmt.Errorf("invalid wal segment path")},
-		{"00000000/", 0, 0, fmt.Errorf("invalid wal segment path")},
-		{"00000000", 0, 0, fmt.Errorf("invalid wal segment path")},
-		{"", 0, 0, fmt.Errorf("invalid wal segment path")},
-	} {
-		t.Run("", func(t *testing.T) {
-			index, offset, err := internal.ParseWALSegmentPath(tt.s)
-			if got, want := index, tt.index; got != want {
-				t.Errorf("index=%#v, want %#v", got, want)
-			} else if got, want := offset, tt.offset; got != want {
-				t.Errorf("offset=%#v, want %#v", got, want)
-			} else if got, want := err, tt.err; !reflect.DeepEqual(got, want) {
-				t.Errorf("err=%#v, want %#v", got, want)
-			}
-		})
-	}
-}

litestream.go

@@ -7,7 +7,9 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"path"
 	"path/filepath"
+	"regexp"
 	"strconv"
 	"strings"
 	"time"
@@ -382,34 +384,134 @@ func IsGenerationName(s string) bool {
 	return true
 }
 
-// FormatIndex formats an index as an 8-character hex value.
-func FormatIndex(index int) string {
-	return fmt.Sprintf("%08x", index)
+// GenerationsPath returns the path to a generation root directory.
+func GenerationsPath(root string) string {
+	return path.Join(root, "generations")
 }
 
-// ParseIndex parses a hex-formatted index into an integer.
-func ParseIndex(s string) (int, error) {
-	v, err := strconv.ParseUint(s, 16, 32)
-	if err != nil {
-		return -1, fmt.Errorf("cannot parse index: %q", s)
+// GenerationPath returns the path to a generation's root directory.
+func GenerationPath(root, generation string) (string, error) {
+	dir := GenerationsPath(root)
+	if generation == "" {
+		return "", fmt.Errorf("generation required")
 	}
-	return int(v), nil
+	return path.Join(dir, generation), nil
 }
 
-// FormatOffset formats an offset as an 8-character hex value.
-func FormatOffset(offset int64) string {
-	return fmt.Sprintf("%08x", offset)
-}
-
-// ParseOffset parses a hex-formatted offset into an integer.
-func ParseOffset(s string) (int64, error) {
-	v, err := strconv.ParseInt(s, 16, 32)
+// SnapshotsPath returns the path to a generation's snapshot directory.
+func SnapshotsPath(root, generation string) (string, error) {
+	dir, err := GenerationPath(root, generation)
 	if err != nil {
-		return -1, fmt.Errorf("cannot parse index: %q", s)
+		return "", err
 	}
-	return v, nil
+	return path.Join(dir, "snapshots"), nil
 }
 
+// SnapshotPath returns the path to an uncompressed snapshot file.
+func SnapshotPath(root, generation string, index int) (string, error) {
+	dir, err := SnapshotsPath(root, generation)
+	if err != nil {
+		return "", err
+	}
+	return path.Join(dir, FormatSnapshotPath(index)), nil
+}
+
+// WALPath returns the path to a generation's WAL directory
+func WALPath(root, generation string) (string, error) {
+	dir, err := GenerationPath(root, generation)
+	if err != nil {
+		return "", err
+	}
+	return path.Join(dir, "wal"), nil
+}
+
+// WALSegmentPath returns the path to a WAL segment file.
+func WALSegmentPath(root, generation string, index int, offset int64) (string, error) {
+	dir, err := WALPath(root, generation)
+	if err != nil {
+		return "", err
+	}
+	return path.Join(dir, FormatWALSegmentPath(index, offset)), nil
+}
+
+// IsSnapshotPath returns true if s is a path to a snapshot file.
+func IsSnapshotPath(s string) bool {
+	return snapshotPathRegex.MatchString(s)
+}
+
+// ParseSnapshotPath returns the index for the snapshot.
+// Returns an error if the path is not a valid snapshot path.
+func ParseSnapshotPath(s string) (index int, err error) {
+	s = filepath.Base(s)
+
+	a := snapshotPathRegex.FindStringSubmatch(s)
+	if a == nil {
+		return 0, fmt.Errorf("invalid snapshot path: %s", s)
+	}
+
+	i64, _ := strconv.ParseUint(a[1], 16, 64)
+	return int(i64), nil
+}
+
+// FormatSnapshotPath formats a snapshot filename with a given index.
+func FormatSnapshotPath(index int) string {
+	assert(index >= 0, "snapshot index must be non-negative")
+	return fmt.Sprintf("%08x%s", index, SnapshotExt)
+}
+
+var snapshotPathRegex = regexp.MustCompile(`^([0-9a-f]{8})\.snapshot\.lz4$`)
+
+// IsWALPath returns true if s is a path to a WAL file.
+func IsWALPath(s string) bool {
+	return walPathRegex.MatchString(s)
+}
+
+// ParseWALPath returns the index for the WAL file.
+// Returns an error if the path is not a valid WAL path.
+func ParseWALPath(s string) (index int, err error) {
+	s = filepath.Base(s)
+
+	a := walPathRegex.FindStringSubmatch(s)
+	if a == nil {
+		return 0, fmt.Errorf("invalid wal path: %s", s)
+	}
+
+	i64, _ := strconv.ParseUint(a[1], 16, 64)
+	return int(i64), nil
+}
+
+// FormatWALPath formats a WAL filename with a given index.
+func FormatWALPath(index int) string {
+	assert(index >= 0, "wal index must be non-negative")
+	return fmt.Sprintf("%08x%s", index, WALExt)
+}
+
+var walPathRegex = regexp.MustCompile(`^([0-9a-f]{8})\.wal$`)
+
+// ParseWALSegmentPath returns the index & offset for the WAL segment file.
+// Returns an error if the path is not a valid wal segment path.
+func ParseWALSegmentPath(s string) (index int, offset int64, err error) {
+	s = filepath.Base(s)
+
+	a := walSegmentPathRegex.FindStringSubmatch(s)
+	if a == nil {
+		return 0, 0, fmt.Errorf("invalid wal segment path: %s", s)
+	}
+
+	i64, _ := strconv.ParseUint(a[1], 16, 64)
+	off64, _ := strconv.ParseUint(a[2], 16, 64)
+	return int(i64), int64(off64), nil
+}
+
+// FormatWALSegmentPath formats a WAL segment filename with a given index & offset.
+func FormatWALSegmentPath(index int, offset int64) string {
+	assert(index >= 0, "wal index must be non-negative")
+	assert(offset >= 0, "wal offset must be non-negative")
+	return fmt.Sprintf("%08x_%08x%s", index, offset, WALSegmentExt)
+}
+
+var walSegmentPathRegex = regexp.MustCompile(`^([0-9a-f]{8})(?:_([0-9a-f]{8}))\.wal\.lz4$`)
+
 // isHexChar returns true if ch is a lowercase hex character.
 func isHexChar(ch rune) bool {
 	return (ch >= '0' && ch <= '9') || (ch >= 'a' && ch <= 'f')

litestream_test.go

@@ -40,6 +40,94 @@ func TestChecksum(t *testing.T) {
 	})
 }
 
+func TestGenerationsPath(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		if got, want := litestream.GenerationsPath("foo"), "foo/generations"; got != want {
+			t.Fatalf("GenerationsPath()=%v, want %v", got, want)
+		}
+	})
+	t.Run("NoPath", func(t *testing.T) {
+		if got, want := litestream.GenerationsPath(""), "generations"; got != want {
+			t.Fatalf("GenerationsPath()=%v, want %v", got, want)
+		}
+	})
+}
+
+func TestGenerationPath(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		if got, err := litestream.GenerationPath("foo", "0123456701234567"); err != nil {
+			t.Fatal(err)
+		} else if want := "foo/generations/0123456701234567"; got != want {
+			t.Fatalf("GenerationPath()=%v, want %v", got, want)
+		}
+	})
+	t.Run("ErrNoGeneration", func(t *testing.T) {
+		if _, err := litestream.GenerationPath("foo", ""); err == nil || err.Error() != `generation required` {
+			t.Fatalf("expected error: %v", err)
+		}
+	})
+}
+
+func TestSnapshotsPath(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		if got, err := litestream.SnapshotsPath("foo", "0123456701234567"); err != nil {
+			t.Fatal(err)
+		} else if want := "foo/generations/0123456701234567/snapshots"; got != want {
+			t.Fatalf("SnapshotsPath()=%v, want %v", got, want)
+		}
+	})
+	t.Run("ErrNoGeneration", func(t *testing.T) {
+		if _, err := litestream.SnapshotsPath("foo", ""); err == nil || err.Error() != `generation required` {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+}
+
+func TestSnapshotPath(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		if got, err := litestream.SnapshotPath("foo", "0123456701234567", 1000); err != nil {
+			t.Fatal(err)
+		} else if want := "foo/generations/0123456701234567/snapshots/000003e8.snapshot.lz4"; got != want {
+			t.Fatalf("SnapshotPath()=%v, want %v", got, want)
+		}
+	})
+	t.Run("ErrNoGeneration", func(t *testing.T) {
+		if _, err := litestream.SnapshotPath("foo", "", 1000); err == nil || err.Error() != `generation required` {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+}
+
+func TestWALPath(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		if got, err := litestream.WALPath("foo", "0123456701234567"); err != nil {
+			t.Fatal(err)
+		} else if want := "foo/generations/0123456701234567/wal"; got != want {
+			t.Fatalf("WALPath()=%v, want %v", got, want)
+		}
+	})
+	t.Run("ErrNoGeneration", func(t *testing.T) {
+		if _, err := litestream.WALPath("foo", ""); err == nil || err.Error() != `generation required` {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+}
+
+func TestWALSegmentPath(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		if got, err := litestream.WALSegmentPath("foo", "0123456701234567", 1000, 1001); err != nil {
+			t.Fatal(err)
+		} else if want := "foo/generations/0123456701234567/wal/000003e8_000003e9.wal.lz4"; got != want {
+			t.Fatalf("WALPath()=%v, want %v", got, want)
+		}
+	})
+	t.Run("ErrNoGeneration", func(t *testing.T) {
+		if _, err := litestream.WALSegmentPath("foo", "", 1000, 0); err == nil || err.Error() != `generation required` {
+			t.Fatalf("unexpected error: %v", err)
+		}
+	})
+}
+
 func MustDecodeHexString(s string) []byte {
 	b, err := hex.DecodeString(s)
 	if err != nil {

replica.go

@@ -159,22 +159,21 @@ func (r *Replica) Sync(ctx context.Context) (err error) {
 
 	Tracef("%s(%s): replica sync: db.pos=%s", r.db.Path(), r.Name(), dpos)
 
-	// Create snapshot if no snapshots exist for generation.
-	snapshotN, err := r.snapshotN(generation)
-	if err != nil {
-		return err
-	} else if snapshotN == 0 {
-		if info, err := r.Snapshot(ctx); err != nil {
-			return err
-		} else if info.Generation != generation {
-			return fmt.Errorf("generation changed during snapshot, exiting sync")
-		}
-		snapshotN = 1
-	}
-	replicaSnapshotTotalGaugeVec.WithLabelValues(r.db.Path(), r.Name()).Set(float64(snapshotN))
-
-	// Determine position, if necessary.
+	// Create a new snapshot and update the current replica position if
+	// the generation on the database has changed.
 	if r.Pos().Generation != generation {
+		// Create snapshot if no snapshots exist for generation.
+		snapshotN, err := r.snapshotN(generation)
+		if err != nil {
+			return err
+		} else if snapshotN == 0 {
+			if info, err := r.Snapshot(ctx); err != nil {
+				return err
+			} else if info.Generation != generation {
+				return fmt.Errorf("generation changed during snapshot, exiting sync")
+			}
+		}
+
 		pos, err := r.calcPos(ctx, generation)
 		if err != nil {
 			return fmt.Errorf("cannot determine replica position: %s", err)
@@ -1307,13 +1306,6 @@ func (r *Replica) downloadWAL(ctx context.Context, generation string, index int,
 
 // Replica metrics.
 var (
-	replicaSnapshotTotalGaugeVec = promauto.NewGaugeVec(prometheus.GaugeOpts{
-		Namespace: "litestream",
-		Subsystem: "replica",
-		Name:      "snapshot_total",
-		Help:      "The current number of snapshots",
-	}, []string{"db", "name"})
-
 	replicaWALBytesCounterVec = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: "litestream",
 		Subsystem: "replica",

replica_client_test.go

@@ -177,7 +177,7 @@ func TestReplicaClient_Snapshots(t *testing.T) {
 		if err == nil {
 			err = itr.Close()
 		}
-		if err == nil || err.Error() != `generation required` {
+		if err == nil || err.Error() != `cannot determine snapshots path: generation required` {
 			t.Fatalf("unexpected error: %v", err)
 		}
 	})
@@ -204,7 +204,7 @@ func TestReplicaClient_WriteSnapshot(t *testing.T) {
 
 	RunWithReplicaClient(t, "ErrNoGeneration", func(t *testing.T, c litestream.ReplicaClient) {
 		t.Parallel()
-		if _, err := c.WriteSnapshot(context.Background(), "", 0, nil); err == nil || err.Error() != `generation required` {
+		if _, err := c.WriteSnapshot(context.Background(), "", 0, nil); err == nil || err.Error() != `cannot determine snapshot path: generation required` {
 			t.Fatalf("unexpected error: %v", err)
 		}
 	})
@@ -242,13 +242,13 @@ func TestReplicaClient_SnapshotReader(t *testing.T) {
 	RunWithReplicaClient(t, "ErrNoGeneration", func(t *testing.T, c litestream.ReplicaClient) {
 		t.Parallel()
 
-		if _, err := c.SnapshotReader(context.Background(), "", 1); err == nil || err.Error() != `generation required` {
+		if _, err := c.SnapshotReader(context.Background(), "", 1); err == nil || err.Error() != `cannot determine snapshot path: generation required` {
 			t.Fatalf("unexpected error: %v", err)
 		}
 	})
 }
 
-func TestReplicaClient_WALSegments(t *testing.T) {
+func TestReplicaClient_WALs(t *testing.T) {
 	RunWithReplicaClient(t, "OK", func(t *testing.T, c litestream.ReplicaClient) {
 		t.Parallel()
 
@@ -362,7 +362,7 @@ func TestReplicaClient_WALSegments(t *testing.T) {
 		if err == nil {
 			err = itr.Close()
 		}
-		if err == nil || err.Error() != `generation required` {
+		if err == nil || err.Error() != `cannot determine wal path: generation required` {
 			t.Fatalf("unexpected error: %v", err)
 		}
 	})
@@ -389,13 +389,13 @@ func TestReplicaClient_WriteWALSegment(t *testing.T) {
 
 	RunWithReplicaClient(t, "ErrNoGeneration", func(t *testing.T, c litestream.ReplicaClient) {
 		t.Parallel()
-		if _, err := c.WriteWALSegment(context.Background(), litestream.Pos{Generation: "", Index: 0, Offset: 0}, nil); err == nil || err.Error() != `generation required` {
+		if _, err := c.WriteWALSegment(context.Background(), litestream.Pos{Generation: "", Index: 0, Offset: 0}, nil); err == nil || err.Error() != `cannot determine wal segment path: generation required` {
 			t.Fatalf("unexpected error: %v", err)
 		}
 	})
 }
 
-func TestReplicaClient_WALSegmentReader(t *testing.T) {
+func TestReplicaClient_WALReader(t *testing.T) {
 
 	RunWithReplicaClient(t, "OK", func(t *testing.T, c litestream.ReplicaClient) {
 		t.Parallel()
@@ -451,7 +451,7 @@ func TestReplicaClient_DeleteWALSegments(t *testing.T) {
 
 	RunWithReplicaClient(t, "ErrNoGeneration", func(t *testing.T, c litestream.ReplicaClient) {
 		t.Parallel()
-		if err := c.DeleteWALSegments(context.Background(), []litestream.Pos{{}}); err == nil || err.Error() != `generation required` {
+		if err := c.DeleteWALSegments(context.Background(), []litestream.Pos{{}}); err == nil || err.Error() != `cannot determine wal segment path: generation required` {
 			t.Fatalf("unexpected error: %v", err)
 		}
 	})

s3/replica_client.go

@@ -10,7 +10,6 @@ import (
 	"os"
 	"path"
 	"regexp"
-	"strings"
 	"sync"
 	"time"
 
@@ -155,7 +154,7 @@ func (c *ReplicaClient) Generations(ctx context.Context) ([]string, error) {
 	var generations []string
 	if err := c.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{
 		Bucket:    aws.String(c.Bucket),
-		Prefix:    aws.String(path.Join(c.Path, "generations") + "/"),
+		Prefix:    aws.String(litestream.GenerationsPath(c.Path) + "/"),
 		Delimiter: aws.String("/"),
 	}, func(page *s3.ListObjectsOutput, lastPage bool) bool {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
@@ -179,15 +178,18 @@ func (c *ReplicaClient) Generations(ctx context.Context) ([]string, error) {
 func (c *ReplicaClient) DeleteGeneration(ctx context.Context, generation string) error {
 	if err := c.Init(ctx); err != nil {
 		return err
-	} else if generation == "" {
-		return fmt.Errorf("generation required")
+	}
+
+	dir, err := litestream.GenerationPath(c.Path, generation)
+	if err != nil {
+		return fmt.Errorf("cannot determine generation path: %w", err)
 	}
 
 	// Collect all files for the generation.
 	var objIDs []*s3.ObjectIdentifier
 	if err := c.s3.ListObjectsPagesWithContext(ctx, &s3.ListObjectsInput{
 		Bucket: aws.String(c.Bucket),
-		Prefix: aws.String(path.Join(c.Path, "generations", generation)),
+		Prefix: aws.String(dir),
 	}, func(page *s3.ListObjectsOutput, lastPage bool) bool {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
@@ -234,11 +236,12 @@ func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (lites
 func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, index int, rd io.Reader) (info litestream.SnapshotInfo, err error) {
 	if err := c.Init(ctx); err != nil {
 		return info, err
-	} else if generation == "" {
-		return info, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	key, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return info, fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 	startTime := time.Now()
 
 	rc := internal.NewReadCounter(rd)
@@ -267,11 +270,12 @@ func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, in
 func (c *ReplicaClient) SnapshotReader(ctx context.Context, generation string, index int) (io.ReadCloser, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
-	} else if generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	key, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 
 	out, err := c.s3.GetObjectWithContext(ctx, &s3.GetObjectInput{
 		Bucket: aws.String(c.Bucket),
@@ -292,11 +296,12 @@ func (c *ReplicaClient) SnapshotReader(ctx context.Context, generation string, i
 func (c *ReplicaClient) DeleteSnapshot(ctx context.Context, generation string, index int) error {
 	if err := c.Init(ctx); err != nil {
 		return err
-	} else if generation == "" {
-		return fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	key, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 
 	if _, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
 		Bucket: aws.String(c.Bucket),
@@ -321,11 +326,12 @@ func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (lit
 func (c *ReplicaClient) WriteWALSegment(ctx context.Context, pos litestream.Pos, rd io.Reader) (info litestream.WALSegmentInfo, err error) {
 	if err := c.Init(ctx); err != nil {
 		return info, err
-	} else if pos.Generation == "" {
-		return info, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
+	key, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+	if err != nil {
+		return info, fmt.Errorf("cannot determine wal segment path: %w", err)
+	}
 	startTime := time.Now()
 
 	rc := internal.NewReadCounter(rd)
@@ -354,11 +360,12 @@ func (c *ReplicaClient) WriteWALSegment(ctx context.Context, pos litestream.Pos,
 func (c *ReplicaClient) WALSegmentReader(ctx context.Context, pos litestream.Pos) (io.ReadCloser, error) {
 	if err := c.Init(ctx); err != nil {
 		return nil, err
-	} else if pos.Generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
 
-	key := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
+	key, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine wal segment path: %w", err)
+	}
 
 	out, err := c.s3.GetObjectWithContext(ctx, &s3.GetObjectInput{
 		Bucket: aws.String(c.Bucket),
@@ -390,10 +397,10 @@ func (c *ReplicaClient) DeleteWALSegments(ctx context.Context, a []litestream.Po
 
 		// Generate a batch of object IDs for deleting the WAL segments.
 		for i, pos := range a[:n] {
-			if pos.Generation == "" {
-				return fmt.Errorf("generation required")
+			key, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+			if err != nil {
+				return fmt.Errorf("cannot determine wal segment path: %w", err)
 			}
-			key := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
 			objIDs[i] = &s3.ObjectIdentifier{Key: &key}
 		}
 
@@ -491,12 +498,11 @@ func newSnapshotIterator(ctx context.Context, client *ReplicaClient, generation
 func (itr *snapshotIterator) fetch() error {
 	defer close(itr.ch)
 
-	if itr.generation == "" {
-		return fmt.Errorf("generation required")
+	dir, err := litestream.SnapshotsPath(itr.client.Path, itr.generation)
+	if err != nil {
+		return fmt.Errorf("cannot determine snapshots path: %w", err)
 	}
 
-	dir := path.Join(itr.client.Path, "generations", itr.generation, "snapshots")
-
 	return itr.client.s3.ListObjectsPagesWithContext(itr.ctx, &s3.ListObjectsInput{
 		Bucket:    aws.String(itr.client.Bucket),
 		Prefix:    aws.String(dir + "/"),
@@ -505,7 +511,8 @@ func (itr *snapshotIterator) fetch() error {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
 		for _, obj := range page.Contents {
-			index, err := internal.ParseSnapshotPath(path.Base(*obj.Key))
+			key := path.Base(*obj.Key)
+			index, err := litestream.ParseSnapshotPath(key)
 			if err != nil {
 				continue
 			}
@@ -594,20 +601,21 @@ func newWALSegmentIterator(ctx context.Context, client *ReplicaClient, generatio
 func (itr *walSegmentIterator) fetch() error {
 	defer close(itr.ch)
 
-	if itr.generation == "" {
-		return fmt.Errorf("generation required")
+	dir, err := litestream.WALPath(itr.client.Path, itr.generation)
+	if err != nil {
+		return fmt.Errorf("cannot determine wal path: %w", err)
 	}
 
-	prefix := path.Join(itr.client.Path, "generations", itr.generation, "wal") + "/"
-
 	return itr.client.s3.ListObjectsPagesWithContext(itr.ctx, &s3.ListObjectsInput{
-		Bucket: aws.String(itr.client.Bucket),
-		Prefix: aws.String(prefix),
+		Bucket:    aws.String(itr.client.Bucket),
+		Prefix:    aws.String(dir + "/"),
+		Delimiter: aws.String("/"),
 	}, func(page *s3.ListObjectsOutput, lastPage bool) bool {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
 		for _, obj := range page.Contents {
-			index, offset, err := internal.ParseWALSegmentPath(strings.TrimPrefix(*obj.Key, prefix))
+			key := path.Base(*obj.Key)
+			index, offset, err := litestream.ParseWALSegmentPath(key)
 			if err != nil {
 				continue
 			}

sftp/replica_client.go

@@ -9,7 +9,6 @@ import (
 	"os"
 	"path"
 	"sort"
-	"strings"
 	"sync"
 	"time"
 
@@ -122,7 +121,7 @@ func (c *ReplicaClient) Generations(ctx context.Context) (_ []string, err error)
 		return nil, err
 	}
 
-	fis, err := sftpClient.ReadDir(path.Join(c.Path, "generations"))
+	fis, err := sftpClient.ReadDir(litestream.GenerationsPath(c.Path))
 	if os.IsNotExist(err) {
 		return nil, nil
 	} else if err != nil {
@@ -154,11 +153,12 @@ func (c *ReplicaClient) DeleteGeneration(ctx context.Context, generation string)
 	sftpClient, err := c.Init(ctx)
 	if err != nil {
 		return err
-	} else if generation == "" {
-		return fmt.Errorf("generation required")
 	}
 
-	dir := path.Join(c.Path, "generations", generation)
+	dir, err := litestream.GenerationPath(c.Path, generation)
+	if err != nil {
+		return fmt.Errorf("cannot determine generation path: %w", err)
+	}
 
 	var dirs []string
 	walker := sftpClient.Walk(dir)
@@ -198,11 +198,12 @@ func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (_ lit
 	sftpClient, err := c.Init(ctx)
 	if err != nil {
 		return nil, err
-	} else if generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
 
-	dir := path.Join(c.Path, "generations", generation, "snapshots")
+	dir, err := litestream.SnapshotsPath(c.Path, generation)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine snapshots path: %w", err)
+	}
 
 	fis, err := sftpClient.ReadDir(dir)
 	if os.IsNotExist(err) {
@@ -215,7 +216,7 @@ func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (_ lit
 	infos := make([]litestream.SnapshotInfo, 0, len(fis))
 	for _, fi := range fis {
 		// Parse index from filename.
-		index, err := internal.ParseSnapshotPath(path.Base(fi.Name()))
+		index, err := litestream.ParseSnapshotPath(path.Base(fi.Name()))
 		if err != nil {
 			continue
 		}
@@ -240,11 +241,12 @@ func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, in
 	sftpClient, err := c.Init(ctx)
 	if err != nil {
 		return info, err
-	} else if generation == "" {
-		return info, fmt.Errorf("generation required")
 	}
 
-	filename := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	filename, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return info, fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 	startTime := time.Now()
 
 	if err := sftpClient.MkdirAll(path.Dir(filename)); err != nil {
@@ -284,11 +286,12 @@ func (c *ReplicaClient) SnapshotReader(ctx context.Context, generation string, i
 	sftpClient, err := c.Init(ctx)
 	if err != nil {
 		return nil, err
-	} else if generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
 
-	filename := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	filename, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 
 	f, err := sftpClient.Open(filename)
 	if err != nil {
@@ -307,11 +310,12 @@ func (c *ReplicaClient) DeleteSnapshot(ctx context.Context, generation string, i
 	sftpClient, err := c.Init(ctx)
 	if err != nil {
 		return err
-	} else if generation == "" {
-		return fmt.Errorf("generation required")
 	}
 
-	filename := path.Join(c.Path, "generations", generation, "snapshots", litestream.FormatIndex(index)+".snapshot.lz4")
+	filename, err := litestream.SnapshotPath(c.Path, generation, index)
+	if err != nil {
+		return fmt.Errorf("cannot determine snapshot path: %w", err)
+	}
 
 	if err := sftpClient.Remove(filename); err != nil && !os.IsNotExist(err) {
 		return fmt.Errorf("cannot delete snapshot %q: %w", filename, err)
@@ -328,11 +332,12 @@ func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (_ l
 	sftpClient, err := c.Init(ctx)
 	if err != nil {
 		return nil, err
-	} else if generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
 
-	dir := path.Join(c.Path, "generations", generation, "wal")
+	dir, err := litestream.WALPath(c.Path, generation)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine wal path: %w", err)
+	}
 
 	fis, err := sftpClient.ReadDir(dir)
 	if os.IsNotExist(err) {
@@ -342,18 +347,25 @@ func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (_ l
 	}
 
 	// Iterate over every file and convert to metadata.
-	indexes := make([]int, 0, len(fis))
+	infos := make([]litestream.WALSegmentInfo, 0, len(fis))
 	for _, fi := range fis {
-		index, err := litestream.ParseIndex(fi.Name())
-		if err != nil || !fi.IsDir() {
+		index, offset, err := litestream.ParseWALSegmentPath(path.Base(fi.Name()))
+		if err != nil {
 			continue
 		}
-		indexes = append(indexes, index)
+
+		infos = append(infos, litestream.WALSegmentInfo{
+			Generation: generation,
+			Index:      index,
+			Offset:     offset,
+			Size:       fi.Size(),
+			CreatedAt:  fi.ModTime().UTC(),
+		})
 	}
 
-	sort.Ints(indexes)
+	sort.Sort(litestream.WALSegmentInfoSlice(infos))
 
-	return newWALSegmentIterator(ctx, c, dir, generation, indexes), nil
+	return litestream.NewWALSegmentInfoSliceIterator(infos), nil
 }
 
 // WriteWALSegment writes LZ4 compressed data from rd into a file on disk.
@@ -363,11 +375,12 @@ func (c *ReplicaClient) WriteWALSegment(ctx context.Context, pos litestream.Pos,
 	sftpClient, err := c.Init(ctx)
 	if err != nil {
 		return info, err
-	} else if pos.Generation == "" {
-		return info, fmt.Errorf("generation required")
 	}
 
-	filename := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
+	filename, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+	if err != nil {
+		return info, fmt.Errorf("cannot determine wal segment path: %w", err)
+	}
 	startTime := time.Now()
 
 	if err := sftpClient.MkdirAll(path.Dir(filename)); err != nil {
@@ -407,11 +420,12 @@ func (c *ReplicaClient) WALSegmentReader(ctx context.Context, pos litestream.Pos
 	sftpClient, err := c.Init(ctx)
 	if err != nil {
 		return nil, err
-	} else if pos.Generation == "" {
-		return nil, fmt.Errorf("generation required")
 	}
 
-	filename := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
+	filename, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+	if err != nil {
+		return nil, fmt.Errorf("cannot determine wal segment path: %w", err)
+	}
 
 	f, err := sftpClient.Open(filename)
 	if err != nil {
@@ -433,12 +447,11 @@ func (c *ReplicaClient) DeleteWALSegments(ctx context.Context, a []litestream.Po
 	}
 
 	for _, pos := range a {
-		if pos.Generation == "" {
-			return fmt.Errorf("generation required")
+		filename, err := litestream.WALSegmentPath(c.Path, pos.Generation, pos.Index, pos.Offset)
+		if err != nil {
+			return fmt.Errorf("cannot determine wal segment path: %w", err)
 		}
 
-		filename := path.Join(c.Path, "generations", pos.Generation, "wal", litestream.FormatIndex(pos.Index), litestream.FormatOffset(pos.Offset)+".wal.lz4")
-
 		if err := sftpClient.Remove(filename); err != nil && !os.IsNotExist(err) {
 			return fmt.Errorf("cannot delete wal segment %q: %w", filename, err)
 		}
@@ -457,7 +470,7 @@ func (c *ReplicaClient) Cleanup(ctx context.Context) (err error) {
 		return err
 	}
 
-	if err := sftpClient.RemoveDirectory(path.Join(c.Path, "generations")); err != nil && !os.IsNotExist(err) {
+	if err := sftpClient.RemoveDirectory(litestream.GenerationsPath(c.Path)); err != nil && !os.IsNotExist(err) {
 		return fmt.Errorf("cannot delete generations path: %w", err)
 	} else if err := sftpClient.RemoveDirectory(c.Path); err != nil && !os.IsNotExist(err) {
 		return fmt.Errorf("cannot delete path: %w", err)
@@ -480,101 +493,3 @@ func (c *ReplicaClient) resetOnConnError(err error) {
 		c.sshClient = nil
 	}
 }
-
-type walSegmentIterator struct {
-	ctx        context.Context
-	client     *ReplicaClient
-	dir        string
-	generation string
-	indexes    []int
-
-	infos []litestream.WALSegmentInfo
-	err   error
-}
-
-func newWALSegmentIterator(ctx context.Context, client *ReplicaClient, dir, generation string, indexes []int) *walSegmentIterator {
-	return &walSegmentIterator{
-		ctx:        ctx,
-		client:     client,
-		dir:        dir,
-		generation: generation,
-		indexes:    indexes,
-	}
-}
-
-func (itr *walSegmentIterator) Close() (err error) {
-	return itr.err
-}
-
-func (itr *walSegmentIterator) Next() bool {
-	sftpClient, err := itr.client.Init(itr.ctx)
-	if err != nil {
-		itr.err = err
-		return false
-	}
-
-	// Exit if an error has already occurred.
-	if itr.err != nil {
-		return false
-	}
-
-	for {
-		// Move to the next segment in cache, if available.
-		if len(itr.infos) > 1 {
-			itr.infos = itr.infos[1:]
-			return true
-		}
-		itr.infos = itr.infos[:0] // otherwise clear infos
-
-		// Move to the next index unless this is the first time initializing.
-		if itr.infos != nil && len(itr.indexes) > 0 {
-			itr.indexes = itr.indexes[1:]
-		}
-
-		// If no indexes remain, stop iteration.
-		if len(itr.indexes) == 0 {
-			return false
-		}
-
-		// Read segments into a cache for the current index.
-		index := itr.indexes[0]
-		fis, err := sftpClient.ReadDir(path.Join(itr.dir, litestream.FormatIndex(index)))
-		if err != nil {
-			itr.err = err
-			return false
-		}
-
-		for _, fi := range fis {
-			filename := path.Base(fi.Name())
-			if fi.IsDir() {
-				continue
-			}
-
-			offset, err := litestream.ParseOffset(strings.TrimSuffix(filename, ".wal.lz4"))
-			if err != nil {
-				continue
-			}
-
-			itr.infos = append(itr.infos, litestream.WALSegmentInfo{
-				Generation: itr.generation,
-				Index:      index,
-				Offset:     offset,
-				Size:       fi.Size(),
-				CreatedAt:  fi.ModTime().UTC(),
-			})
-		}
-
-		if len(itr.infos) > 0 {
-			return true
-		}
-	}
-}
-
-func (itr *walSegmentIterator) Err() error { return itr.err }
-
-func (itr *walSegmentIterator) WALSegment() litestream.WALSegmentInfo {
-	if len(itr.infos) == 0 {
-		return litestream.WALSegmentInfo{}
-	}
-	return itr.infos[0]
-}