func (r *Replica) retainer(ctx context.Context) {
	if r.Retention <= 0 {
		return
	}
	checkInterval := r.RetentionCheckInterval
	if checkInterval > r.Retention {
		checkInterval = r.Retention
	}
	ticker := time.NewTicker(checkInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			if err := r.EnforceRetention(ctx); err != nil {
				r.Logger.Printf("retainer error: %s", err)
				continue
			}
		}
	}
}
retainer runs in a separate goroutine and handles retention.
package litestream
import (
	"context"
	"fmt"
	"hash/crc64"
	"io"
	"io/ioutil"
	"log"
	"math"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"
	"github.com/benbjohnson/litestream/internal"
	"github.com/pierrec/lz4/v4"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"golang.org/x/sync/errgroup"
)
const (
	DefaultSyncInterval           = 1 * time.Second
	DefaultRetention              = 24 * time.Hour
	DefaultRetentionCheckInterval = 1 * time.Hour
)
type Replica struct {
	db   *DB
	name string
	mu  sync.RWMutex
	pos Pos  
	muf sync.Mutex
	f   *os.File  
	wg     sync.WaitGroup
	cancel func()
	Client ReplicaClient
	SyncInterval time.Duration
	SnapshotInterval time.Duration
	Retention time.Duration
	RetentionCheckInterval time.Duration
	ValidationInterval time.Duration
	MonitorEnabled bool
	Logger *log.Logger
}
func NewReplica(db *DB, name string) *Replica {
	r := &Replica{
		db:     db,
		name:   name,
		cancel: func() {},
		SyncInterval:           DefaultSyncInterval,
		Retention:              DefaultRetention,
		RetentionCheckInterval: DefaultRetentionCheckInterval,
		MonitorEnabled:         true,
	}
	prefix := fmt.Sprintf("%s: ", r.Name())
	if db != nil {
		prefix = fmt.Sprintf("%s(%s): ", db.Path(), r.Name())
	}
	r.Logger = log.New(LogWriter, prefix, LogFlags)
	return r
}
func (r *Replica) Name() string {
	if r.name == "" && r.Client != nil {
		return r.Client.Type()
	}
	return r.name
}
func (r *Replica) DB() *DB { return r.db }
func (r *Replica) Start(ctx context.Context) error {
	if !r.MonitorEnabled {
		return nil
	}
	r.Stop(false)
	ctx, r.cancel = context.WithCancel(ctx)
	r.wg.Add(4)
	go func() { defer r.wg.Done(); r.monitor(ctx) }()
	go func() { defer r.wg.Done(); r.retainer(ctx) }()
	go func() { defer r.wg.Done(); r.snapshotter(ctx) }()
	go func() { defer r.wg.Done(); r.validator(ctx) }()
	return nil
}
func (r *Replica) Stop(hard bool) (err error) {
	r.cancel()
	r.wg.Wait()
	r.muf.Lock()
	defer r.muf.Unlock()
	if hard && r.f != nil {
		if e := r.f.Close(); e != nil && err == nil {
			err = e
		}
	}
	return err
}
func (r *Replica) Sync(ctx context.Context) (err error) {
	defer func() {
		if err != nil {
			r.mu.Lock()
			r.pos = Pos{}
			r.mu.Unlock()
		}
	}()
	dpos := r.db.Pos()
	if dpos.IsZero() {
		return fmt.Errorf("no generation, waiting for data")
	}
	generation := dpos.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))
	if r.Pos().Generation != generation {
		pos, err := r.calcPos(ctx, generation)
		if err != nil {
			return fmt.Errorf("cannot determine replica position: %s", err)
		}
		r.mu.Lock()
		r.pos = pos
		r.mu.Unlock()
	}
	if err = r.syncWAL(ctx); err != nil {
		return err
	}
	return nil
}
func (r *Replica) syncWAL(ctx context.Context) (err error) {
	pos := r.Pos()
	itr, err := r.db.WALSegments(ctx, pos.Generation)
	if err != nil {
		return err
	}
	defer itr.Close()
	var segments [][]WALSegmentInfo
	for itr.Next() {
		info := itr.WALSegment()
		if cmp, err := ComparePos(pos, info.Pos()); err != nil {
			return fmt.Errorf("compare pos: %w", err)
		} else if cmp == 1 {
			continue  
		}
		if len(segments) == 0 || segments[len(segments)-1][0].Index != info.Index {
			segments = append(segments, []WALSegmentInfo{info})
			continue
		}
		segments[len(segments)-1] = append(segments[len(segments)-1], info)
	}
	for i := range segments {
		if err := r.writeIndexSegments(ctx, segments[i]); err != nil {
			return fmt.Errorf("write index segments: index=%d err=%w", segments[i][0].Index, err)
		}
	}
	return nil
}
func (r *Replica) writeIndexSegments(ctx context.Context, segments []WALSegmentInfo) (err error) {
	assert(len(segments) > 0, "segments required for replication")
	if pos := r.Pos(); pos != segments[0].Pos() {
		nextIndexPos := pos.Truncate()
		nextIndexPos.Index++
		if nextIndexPos != segments[0].Pos() {
			return fmt.Errorf("replica skipped position: replica=%s initial=%s", pos, segments[0].Pos())
		}
	}
	pos := segments[0].Pos()
	initialPos := pos
	pr, pw := io.Pipe()
	defer func() { _ = pw.CloseWithError(err) }()
	var g errgroup.Group
	g.Go(func() error {
		_, err := r.Client.WriteWALSegment(ctx, initialPos, pr)
		return err
	})
	zw := lz4.NewWriter(pw)
	for _, info := range segments {
		if err := func() error {
			if pos != info.Pos() {
				return fmt.Errorf("non-contiguous segment: expected=%s current=%s", pos, info.Pos())
			}
			rc, err := r.db.WALSegmentReader(ctx, info.Pos())
			if err != nil {
				return err
			}
			defer rc.Close()
			n, err := io.Copy(zw, lz4.NewReader(rc))
			if err != nil {
				return err
			} else if err := rc.Close(); err != nil {
				return err
			}
			pos = info.Pos()
			pos.Offset += n
			return nil
		}(); err != nil {
			return fmt.Errorf("wal segment: pos=%s err=%w", info.Pos(), err)
		}
	}
	if err := zw.Close(); err != nil {
		return err
	} else if err := pw.Close(); err != nil {
		return err
	} else if err := g.Wait(); err != nil {
		return err
	}
	r.mu.Lock()
	r.pos = pos
	r.mu.Unlock()
	replicaWALBytesCounterVec.WithLabelValues(r.db.Path(), r.Name()).Add(float64(pos.Offset - initialPos.Offset))
	replicaWALIndexGaugeVec.WithLabelValues(r.db.Path(), r.Name()).Set(float64(pos.Index))
	replicaWALOffsetGaugeVec.WithLabelValues(r.db.Path(), r.Name()).Set(float64(pos.Offset))
	r.Logger.Printf("wal segment written: %s sz=%d", initialPos, pos.Offset-initialPos.Offset)
	return nil
}
func (r *Replica) snapshotN(generation string) (int, error) {
	itr, err := r.Client.Snapshots(context.Background(), generation)
	if err != nil {
		return 0, err
	}
	defer itr.Close()
	var n int
	for itr.Next() {
		n++
	}
	return n, itr.Close()
}
func (r *Replica) calcPos(ctx context.Context, generation string) (pos Pos, err error) {
	snapshot, err := r.maxSnapshot(ctx, generation)
	if err != nil {
		return pos, fmt.Errorf("max snapshot: %w", err)
	} else if snapshot == nil {
		return pos, fmt.Errorf("no snapshot available: generation=%s", generation)
	}
	segment, err := r.maxWALSegment(ctx, generation)
	if err != nil {
		return pos, fmt.Errorf("max wal segment: %w", err)
	} else if segment == nil {
		return Pos{Generation: snapshot.Generation, Index: snapshot.Index}, nil
	}
	rd, err := r.Client.WALSegmentReader(ctx, segment.Pos())
	if err != nil {
		return pos, fmt.Errorf("wal segment reader: %w", err)
	}
	defer rd.Close()
	n, err := io.Copy(ioutil.Discard, lz4.NewReader(rd))
	if err != nil {
		return pos, err
	}
	return Pos{
		Generation: segment.Generation,
		Index:      segment.Index,
		Offset:     segment.Offset + n,
	}, nil
}
func (r *Replica) maxSnapshot(ctx context.Context, generation string) (*SnapshotInfo, error) {
	itr, err := r.Client.Snapshots(ctx, generation)
	if err != nil {
		return nil, err
	}
	defer itr.Close()
	var max *SnapshotInfo
	for itr.Next() {
		if info := itr.Snapshot(); max == nil || info.Index > max.Index {
			max = &info
		}
	}
	return max, itr.Close()
}
func (r *Replica) maxWALSegment(ctx context.Context, generation string) (*WALSegmentInfo, error) {
	itr, err := r.Client.WALSegments(ctx, generation)
	if err != nil {
		return nil, err
	}
	defer itr.Close()
	var max *WALSegmentInfo
	for itr.Next() {
		if info := itr.WALSegment(); max == nil || info.Index > max.Index || (info.Index == max.Index && info.Offset > max.Offset) {
			max = &info
		}
	}
	return max, itr.Close()
}
func (r *Replica) Pos() Pos {
	r.mu.RLock()
	defer r.mu.RUnlock()
	return r.pos
}
func (r *Replica) Snapshots(ctx context.Context) ([]SnapshotInfo, error) {
	generations, err := r.Client.Generations(ctx)
	if err != nil {
		return nil, fmt.Errorf("cannot fetch generations: %w", err)
	}
	var a []SnapshotInfo
	for _, generation := range generations {
		if err := func() error {
			itr, err := r.Client.Snapshots(ctx, generation)
			if err != nil {
				return err
			}
			defer itr.Close()
			other, err := SliceSnapshotIterator(itr)
			if err != nil {
				return err
			}
			a = append(a, other...)
			return itr.Close()
		}(); err != nil {
			return a, err
		}
	}
	sort.Sort(SnapshotInfoSlice(a))
	return a, nil
}
func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
	if r.db == nil || r.db.db == nil {
		return info, fmt.Errorf("no database available")
	}
	r.muf.Lock()
	defer r.muf.Unlock()
	if _, err := r.db.db.ExecContext(ctx, `PRAGMA wal_checkpoint(PASSIVE);`); err != nil {
		return info, fmt.Errorf("pre-snapshot checkpoint: %w", err)
	}
	tx, err := r.db.db.Begin()
	if err != nil {
		return info, err
	} else if _, err := tx.ExecContext(ctx, `SELECT COUNT(1) FROM _litestream_seq;`); err != nil {
		_ = tx.Rollback()
		return info, err
	}
	defer func() { _ = tx.Rollback() }()
	pos := r.db.Pos()
	if pos.IsZero() {
		return info, ErrNoGeneration
	}
	if r.f == nil {
		if r.f, err = os.Open(r.db.Path()); err != nil {
			return info, err
		}
	}
	if _, err := r.f.Seek(0, io.SeekStart); err != nil {
		return info, err
	}
	pr, pw := io.Pipe()
	var g errgroup.Group
	g.Go(func() error {
		zr := lz4.NewWriter(pw)
		defer zr.Close()
		if _, err := io.Copy(zr, r.f); err != nil {
			pw.CloseWithError(err)
			return err
		} else if err := zr.Close(); err != nil {
			pw.CloseWithError(err)
			return err
		}
		return pw.Close()
	})
	if info, err = r.Client.WriteSnapshot(ctx, pos.Generation, pos.Index, pr); err != nil {
		return info, err
	} else if err := g.Wait(); err != nil {
		return info, err
	}
	r.Logger.Printf("snapshot written %s/%08x", pos.Generation, pos.Index)
	return info, nil
}
func (r *Replica) EnforceRetention(ctx context.Context) (err error) {
	snapshots, err := r.Snapshots(ctx)
	if err != nil {
		return fmt.Errorf("snapshots: %w", err)
	}
	retained := FilterSnapshotsAfter(snapshots, time.Now().Add(-r.Retention))
	if len(retained) == 0 {
		snapshot, err := r.Snapshot(ctx)
		if err != nil {
			return fmt.Errorf("snapshot: %w", err)
		}
		retained = append(retained, snapshot)
	}
	generations, err := r.Client.Generations(ctx)
	if err != nil {
		return fmt.Errorf("generations: %w", err)
	}
	for _, generation := range generations {
		snapshot := FindMinSnapshotByGeneration(retained, generation)
		if snapshot == nil {
			if err := r.Client.DeleteGeneration(ctx, generation); err != nil {
				return fmt.Errorf("delete generation: %w", err)
			}
			continue
		}
		if err := r.deleteSnapshotsBeforeIndex(ctx, generation, snapshot.Index); err != nil {
			return fmt.Errorf("delete snapshots before index: %w", err)
		} else if err := r.deleteWALSegmentsBeforeIndex(ctx, generation, snapshot.Index); err != nil {
			return fmt.Errorf("delete wal segments before index: %w", err)
		}
	}
	return nil
}
func (r *Replica) deleteSnapshotsBeforeIndex(ctx context.Context, generation string, index int) error {
	itr, err := r.Client.Snapshots(ctx, generation)
	if err != nil {
		return fmt.Errorf("fetch snapshots: %w", err)
	}
	defer itr.Close()
	for itr.Next() {
		info := itr.Snapshot()
		if info.Index >= index {
			continue
		}
		if err := r.Client.DeleteSnapshot(ctx, info.Generation, info.Index); err != nil {
			return fmt.Errorf("delete snapshot %s/%08x: %w", info.Generation, info.Index, err)
		}
		r.Logger.Printf("snapshot deleted %s/%08x", generation, index)
	}
	return itr.Close()
}
func (r *Replica) deleteWALSegmentsBeforeIndex(ctx context.Context, generation string, index int) error {
	itr, err := r.Client.WALSegments(ctx, generation)
	if err != nil {
		return fmt.Errorf("fetch wal segments: %w", err)
	}
	defer itr.Close()
	var a []Pos
	for itr.Next() {
		info := itr.WALSegment()
		if info.Index >= index {
			continue
		}
		a = append(a, info.Pos())
	}
	if err := itr.Close(); err != nil {
		return err
	}
	if len(a) == 0 {
		return nil
	}
	if err := r.Client.DeleteWALSegments(ctx, a); err != nil {
		return fmt.Errorf("delete wal segments: %w", err)
	}
	for _, pos := range a {
		r.Logger.Printf("wal segmented deleted: %s", pos)
	}
	return nil
}
func (r *Replica) monitor(ctx context.Context) {
	ticker := time.NewTicker(r.SyncInterval)
	defer ticker.Stop()
	ch := make(chan struct{})
	close(ch)
	var notify <-chan struct{} = ch
	for initial := true; ; initial = false {
		if !initial {
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-notify:
		}
		notify = r.db.Notify()
		if err := r.Sync(ctx); err != nil {
			r.Logger.Printf("monitor error: %s", err)
			continue
		}
	}
}