moved metapath

Signed-off-by: rustfix <771054535@qq.com>

.github/CONTRIBUTING.md

@@ -1,17 +1,17 @@
-## Open-source, not open-contribution
+## Contribution Policy
 
-[Similar to SQLite](https://www.sqlite.org/copyright.html), Litestream is open
+Initially, Litestream was closed to outside contributions. The goal was to
-source but closed to contributions. This keeps the code base free of proprietary
+reduce burnout by limiting the maintenance overhead of reviewing and validating
-or licensed code but it also helps me continue to maintain and build Litestream.
+third-party code. However, this policy is overly broad and has prevented small,
+easily testable patches from being contributed.
 
-As the author of [BoltDB](https://github.com/boltdb/bolt), I found that
+Litestream is now open to code contributions for bug fixes only. Features carry
-accepting and maintaining third party patches contributed to my burn out and
+a long-term maintenance burden so they will not be accepted at this time.
-I eventually archived the project. Writing databases & low-level replication
+Please [submit an issue][new-issue] if you have a feature you'd like to
-tools involves nuance and simple one line changes can have profound and
+request.
-unexpected changes in correctness and performance. Small contributions
-typically required hours of my time to properly test and validate them.
 
-I am grateful for community involvement, bug reports, & feature requests. I do
+If you find mistakes in the documentation, please submit a fix to the
-not wish to come off as anything but welcoming, however, I've
+[documentation repository][docs].
-made the decision to keep this project closed to contributions for my own
+
-mental health and long term viability of the project.
+[new-issue]: https://github.com/benbjohnson/litestream/issues/new
+[docs]: https://github.com/benbjohnson/litestream.io

.github/pull_request_template.md

@@ -1,7 +0,0 @@
-Litestream is not accepting code contributions at this time. You can find a summary of why on the project's GitHub README:
-
-https://github.com/benbjohnson/litestream#open-source-not-open-contribution
-
-Web site & Documentation changes, however, are welcome. You can find that repository here:
-
-https://github.com/benbjohnson/litestream.io

.github/workflows/commit.yml

@@ -0,0 +1,229 @@
+on:
+  push:
+  pull_request:
+    types:
+      - opened
+      - synchronize
+      - reopened
+
+env:
+  GO_VERSION: "1.21"
+
+name: Commit
+jobs:
+  lint:
+    name: Lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-go@v4
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - run: |
+          go install golang.org/x/tools/cmd/goimports@latest
+          go install honnef.co/go/tools/cmd/staticcheck@latest
+          export PATH="$HOME/go/bin:$PATH"
+
+      - uses: pre-commit/action@v3.0.0
+
+  build-windows:
+    name: Build Windows
+    runs-on: ubuntu-latest
+    steps:
+      - run: sudo apt-get install -y mingw-w64
+
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-go@v4
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - run: |
+          go build ./cmd/litestream/
+          file ./litestream.exe
+        env:
+          CGO_ENABLED: "1"
+          GOOS: windows
+          GOARCH: amd64
+          CC: x86_64-w64-mingw32-gcc
+
+  build:
+    name: Build & Unit Test
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-go@v4
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - run: go env
+
+      - run: go install ./cmd/litestream
+
+      - run: go test -v ./...
+
+#  long-running-test:
+#    name: Run Long Running Unit Test
+#    runs-on: ubuntu-latest
+#    steps:
+#      - uses: actions/checkout@v2
+#      - uses: actions/setup-go@v2
+#        with:
+#          go-version: '1.20'
+#      - uses: actions/cache@v2
+#        with:
+#          path: ~/go/pkg/mod
+#          key: ${{ inputs.os }}-go-${{ hashFiles('**/go.sum') }}
+#          restore-keys: ${{ inputs.os }}-go-
+#
+#      - run: go install ./cmd/litestream
+#      - run: go test -v -run=TestCmd_Replicate_LongRunning ./integration -long-running-duration 1m
+
+  s3-mock-test:
+    name: Run S3 Mock Tests
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+#         cache: 'pip'
+      - run: pip install moto[s3,server]
+
+      - uses: actions/setup-go@v4
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - run: go env
+
+      - run: go install ./cmd/litestream
+
+      - run: ./etc/s3_mock.py go test -v ./replica_client_test.go -integration s3
+
+  s3-integration-test:
+    name: Run S3 Integration Tests
+    runs-on: ubuntu-latest
+    needs: build
+    if: github.ref == 'refs/heads/main'
+    concurrency:
+      group: integration-test-s3
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-go@v4
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - run: go env
+
+      - run: go install ./cmd/litestream
+
+      - run: go test -v ./replica_client_test.go -integration s3
+        env:
+          LITESTREAM_S3_ACCESS_KEY_ID: ${{ secrets.LITESTREAM_S3_ACCESS_KEY_ID }}
+          LITESTREAM_S3_SECRET_ACCESS_KEY: ${{ secrets.LITESTREAM_S3_SECRET_ACCESS_KEY }}
+          LITESTREAM_S3_REGION: us-east-1
+          LITESTREAM_S3_BUCKET: integration.litestream.io
+
+  gcp-integration-test:
+    name: Run GCP Integration Tests
+    runs-on: ubuntu-latest
+    needs: build
+    if: github.ref == 'refs/heads/main'
+    concurrency:
+      group: integration-test-gcp
+    steps:
+      - name: Extract GCP credentials
+        run: 'echo "$GOOGLE_APPLICATION_CREDENTIALS" > /opt/gcp.json'
+        shell: bash
+        env:
+          GOOGLE_APPLICATION_CREDENTIALS: ${{secrets.GOOGLE_APPLICATION_CREDENTIALS}}
+
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-go@v4
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - run: go env
+
+      - run: go install ./cmd/litestream
+
+      - run: go test -v ./replica_client_test.go -integration gcs
+        env:
+          GOOGLE_APPLICATION_CREDENTIALS:  /opt/gcp.json
+          LITESTREAM_GCS_BUCKET:           integration.litestream.io
+
+  abs-integration-test:
+    name: Run Azure Blob Store Integration Tests
+    runs-on: ubuntu-latest
+    needs: build
+    if: github.ref == 'refs/heads/main'
+    concurrency:
+      group: integration-test-abs
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-go@v4
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - run: go env
+
+      - run: go install ./cmd/litestream
+
+      - run: go test -v ./replica_client_test.go -integration abs
+        env:
+          LITESTREAM_ABS_ACCOUNT_NAME:     ${{ secrets.LITESTREAM_ABS_ACCOUNT_NAME }}
+          LITESTREAM_ABS_ACCOUNT_KEY:      ${{ secrets.LITESTREAM_ABS_ACCOUNT_KEY }}
+          LITESTREAM_ABS_BUCKET:           integration
+
+  sftp-integration-test:
+    name: Run SFTP Integration Tests
+    runs-on: ubuntu-latest
+    needs: build
+    steps:
+      - name: Prepare OpenSSH server
+        run: |-
+          sudo mkdir -p /test/etc/ssh /test/home /run/sshd /test/data/
+          sudo ssh-keygen -t ed25519 -f /test/etc/ssh/id_ed25519_host -N ""
+          sudo ssh-keygen -t ed25519 -f /test/etc/ssh/id_ed25519 -N ""
+          sudo chmod 0600 /test/etc/ssh/id_ed25519_host /test/etc/ssh/id_ed25519
+          sudo chmod 0644 /test/etc/ssh/id_ed25519_host.pub /test/etc/ssh/id_ed25519.pub
+          sudo cp /test/etc/ssh/id_ed25519 /test/id_ed25519
+          sudo chown $USER /test/id_ed25519
+          sudo tee /test/etc/ssh/sshd_config <<EOF
+          Port 2222
+          HostKey /test/etc/ssh/id_ed25519_host
+          AuthorizedKeysFile /test/etc/ssh/id_ed25519.pub
+          AuthenticationMethods publickey
+          Subsystem sftp internal-sftp
+          UsePAM no
+          LogLevel DEBUG
+          EOF
+          sudo /usr/sbin/sshd -e -f /test/etc/ssh/sshd_config -E /test/debug.log
+
+      - name: Test OpenSSH server works with pubkey auth
+        run: ssh -v -i /test/id_ed25519 -o StrictHostKeyChecking=accept-new -p 2222 root@localhost whoami || (sudo cat /test/debug.log && exit 1)
+
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-go@v4
+        with:
+          go-version: ${{ env.GO_VERSION }}
+
+      - run: go env
+
+      - run: go install ./cmd/litestream
+
+      - run: go test -v ./replica_client_test.go -integration sftp
+        env:
+          LITESTREAM_SFTP_HOST:     "localhost:2222"
+          LITESTREAM_SFTP_USER:     "root"
+          LITESTREAM_SFTP_KEY_PATH: /test/id_ed25519
+          LITESTREAM_SFTP_PATH:     /test/data

.github/workflows/release.docker.yml

@@ -0,0 +1,51 @@
+on:
+  release:
+    types:
+      - published
+# pull_request:
+#   types:
+#     - opened
+#     - synchronize
+#     - reopened
+#   branches-ignore:
+#     - "dependabot/**"
+
+name: Release (Docker)
+jobs:
+  docker:
+    runs-on: ubuntu-latest
+    env:
+      PLATFORMS: "linux/amd64,linux/arm64,linux/arm/v7"
+      VERSION: "${{ github.event_name == 'release' && github.event.release.name || github.sha }}"
+
+    steps:
+      - uses: actions/checkout@v2
+      - uses: docker/setup-qemu-action@v1
+      - uses: docker/setup-buildx-action@v1
+
+      - uses: docker/login-action@v1
+        with:
+          username: benbjohnson
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
+      - id: meta
+        uses: docker/metadata-action@v3
+        with:
+          images: litestream/litestream
+          tags: |
+            type=ref,event=branch
+            type=ref,event=pr
+            type=sha
+            type=sha,format=long
+            type=semver,pattern={{version}}
+            type=semver,pattern={{major}}.{{minor}}
+
+      - uses: docker/build-push-action@v2
+        with:
+          context: .
+          push: true
+          platforms: ${{ env.PLATFORMS }}
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+          build-args: |
+            LITESTREAM_VERSION=${{ env.VERSION }}

.github/workflows/release.linux.yml

@@ -6,7 +6,7 @@ on:
 name: release (linux)
 jobs:
   build:
-    runs-on: ubuntu-18.04
+    runs-on: ubuntu-latest
     strategy:
       matrix:
         include:
@@ -31,7 +31,7 @@ jobs:
       - uses: actions/checkout@v2
       - uses: actions/setup-go@v2
         with:
-          go-version: '1.16'
+          go-version: '1.21'
 
       - id: release
         uses: bruceadams/get-release@v1.2.2
@@ -54,7 +54,7 @@ jobs:
           mkdir -p dist
           cp etc/litestream.yml etc/litestream.service dist
           cat etc/nfpm.yml | LITESTREAM_VERSION=${{ steps.release.outputs.tag_name }} envsubst > dist/nfpm.yml
-          CGO_ENABLED=1 go build -ldflags "-s -w -X 'main.Version=${{ steps.release.outputs.tag_name }}'" -o dist/litestream ./cmd/litestream
+          CGO_ENABLED=1 go build -ldflags "-s -w -extldflags "-static" -X 'main.Version=${{ steps.release.outputs.tag_name }}'" -tags osusergo,netgo,sqlite_omit_load_extension -o dist/litestream ./cmd/litestream
 
           cd dist
           tar -czvf litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.tar.gz litestream

.github/workflows/release.linux_static.yml

@@ -1,62 +0,0 @@
-on:
-  release:
-    types:
-      - created
-
-name: release (linux/static)
-jobs:
-  build:
-    runs-on: ubuntu-18.04
-    strategy:
-      matrix:
-        include:
-          - arch: amd64
-            cc:   gcc
-          - arch: arm64
-            cc:   aarch64-linux-gnu-gcc
-          - arch: arm
-            arm:  6
-            cc:   arm-linux-gnueabi-gcc
-          - arch: arm
-            arm:  7
-            cc:   arm-linux-gnueabihf-gcc
-
-    env:
-      GOOS:   linux
-      GOARCH: ${{ matrix.arch }}
-      GOARM:  ${{ matrix.arm }}
-      CC:     ${{ matrix.cc }}
-
-    steps:
-      - uses: actions/checkout@v2
-      - uses: actions/setup-go@v2
-        with:
-          go-version: '1.16'
-
-      - id: release
-        uses: bruceadams/get-release@v1.2.2
-        env:
-          GITHUB_TOKEN: ${{ github.token }}
-
-      - name: Install cross-compilers
-        run: |
-          sudo apt-get update 
-          sudo apt-get install -y gcc-aarch64-linux-gnu gcc-arm-linux-gnueabihf gcc-arm-linux-gnueabi
-
-      - name: Build litestream
-        run: |
-          rm -rf dist
-          mkdir -p dist
-          CGO_ENABLED=1 go build -ldflags "-s -w -extldflags "-static" -X 'main.Version=${{ steps.release.outputs.tag_name }}'" -tags osusergo,netgo,sqlite_omit_load_extension -o dist/litestream ./cmd/litestream
-          cd dist
-          tar -czvf litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}-static.tar.gz litestream
-
-      - name: Upload release tarball
-        uses: actions/upload-release-asset@v1.0.2
-        env:
-          GITHUB_TOKEN: ${{ github.token }}
-        with:
-          upload_url: ${{ steps.release.outputs.upload_url }}
-          asset_path: ./dist/litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}-static.tar.gz
-          asset_name: litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}-static.tar.gz
-          asset_content_type: application/gzip

.github/workflows/test.yml

@@ -1,62 +0,0 @@
-on: push
-name: test
-jobs:
-  test:
-    runs-on: ubuntu-18.04
-    steps:
-      - uses: actions/setup-go@v2
-        with:
-          go-version: '1.16'
-
-      - uses: actions/checkout@v2
-
-      - uses: actions/cache@v2
-        with:
-          path: ~/go/pkg/mod
-          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
-          restore-keys: |
-            ${{ runner.os }}-go-
-
-      - name: Extract GCP credentials
-        run: 'echo "$GOOGLE_APPLICATION_CREDENTIALS" > /opt/gcp.json'
-        shell: bash
-        env:
-          GOOGLE_APPLICATION_CREDENTIALS: ${{secrets.GOOGLE_APPLICATION_CREDENTIALS}}
-
-      - name: Extract SSH key
-        run: 'echo "$LITESTREAM_SFTP_KEY" > /opt/id_ed25519'
-        shell: bash
-        env:
-          LITESTREAM_SFTP_KEY: ${{secrets.LITESTREAM_SFTP_KEY}}
-
-      - name: Run unit tests
-        run: go test -v ./...
-
-      - name: Run aws s3 tests
-        run: go test -v -run=TestReplicaClient . -integration s3
-        env:
-          LITESTREAM_S3_ACCESS_KEY_ID:     ${{ secrets.LITESTREAM_S3_ACCESS_KEY_ID }}
-          LITESTREAM_S3_SECRET_ACCESS_KEY: ${{ secrets.LITESTREAM_S3_SECRET_ACCESS_KEY }}
-          LITESTREAM_S3_REGION:            ${{ secrets.LITESTREAM_S3_REGION }}
-          LITESTREAM_S3_BUCKET:            ${{ secrets.LITESTREAM_S3_BUCKET }}
-
-      - name: Run google cloud storage (gcs) tests
-        run: go test -v -run=TestReplicaClient . -integration gcs
-        env:
-          GOOGLE_APPLICATION_CREDENTIALS:  /opt/gcp.json
-          LITESTREAM_GCS_BUCKET:           ${{ secrets.LITESTREAM_GCS_BUCKET }}
-
-      - name: Run azure blob storage (abs) tests
-        run: go test -v -run=TestReplicaClient . -integration abs
-        env:
-          LITESTREAM_ABS_ACCOUNT_NAME:     ${{ secrets.LITESTREAM_ABS_ACCOUNT_NAME }}
-          LITESTREAM_ABS_ACCOUNT_KEY:      ${{ secrets.LITESTREAM_ABS_ACCOUNT_KEY }}
-          LITESTREAM_ABS_BUCKET:           ${{ secrets.LITESTREAM_ABS_BUCKET }}
-
-      - name: Run sftp tests
-        run: go test -v -run=TestReplicaClient . -integration sftp
-        env:
-          LITESTREAM_SFTP_HOST:     ${{ secrets.LITESTREAM_SFTP_HOST }}
-          LITESTREAM_SFTP_USER:     ${{ secrets.LITESTREAM_SFTP_USER }}
-          LITESTREAM_SFTP_KEY_PATH: /opt/id_ed25519
-          LITESTREAM_SFTP_PATH:     ${{ secrets.LITESTREAM_SFTP_PATH }}

.pre-commit-config.yaml

@@ -0,0 +1,20 @@
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.1.0
+    hooks:
+      - id: trailing-whitespace
+        exclude_types: [markdown]
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-added-large-files
+
+  - repo: https://github.com/tekwizely/pre-commit-golang
+    rev: v1.0.0-beta.5
+    hooks:
+      - id: go-imports-repo
+        args:
+          - "-local"
+          - "github.com/benbjohnson/litestrem"
+          - "-w"
+      - id: go-vet-repo-mod
+      - id: go-staticcheck-repo-mod

Dockerfile

@@ -1,12 +1,16 @@
-FROM golang:1.16 as builder
+FROM golang:1.21.3 as builder
+
 WORKDIR /src/litestream
 COPY . .
+
 ARG LITESTREAM_VERSION=latest
+
 RUN --mount=type=cache,target=/root/.cache/go-build \
 	--mount=type=cache,target=/go/pkg \
 	go build -ldflags "-s -w -X 'main.Version=${LITESTREAM_VERSION}' -extldflags '-static'" -tags osusergo,netgo,sqlite_omit_load_extension -o /usr/local/bin/litestream ./cmd/litestream
 
-FROM alpine
+
+FROM alpine:3.17.2
 COPY --from=builder /usr/local/bin/litestream /usr/local/bin/litestream
 ENTRYPOINT ["/usr/local/bin/litestream"]
 CMD []

Makefile

@@ -20,11 +20,17 @@ ifndef LITESTREAM_VERSION
 	$(error LITESTREAM_VERSION is undefined)
 endif
 	mkdir -p dist
-	go build -v -ldflags "-s -w -X 'main.Version=${LITESTREAM_VERSION}'"  -o dist/litestream ./cmd/litestream
+
+	GOOS=darwin GOARCH=amd64 CC="gcc -target amd64-apple-macos11" CGO_ENABLED=1 go build -v -ldflags "-s -w -X 'main.Version=${LITESTREAM_VERSION}'"  -o dist/litestream ./cmd/litestream
 	gon etc/gon.hcl
 	mv dist/litestream.zip dist/litestream-${LITESTREAM_VERSION}-darwin-amd64.zip
 	openssl dgst -sha256 dist/litestream-${LITESTREAM_VERSION}-darwin-amd64.zip
 
+	GOOS=darwin GOARCH=arm64 CC="gcc -target arm64-apple-macos11" CGO_ENABLED=1 go build -v -ldflags "-s -w -X 'main.Version=${LITESTREAM_VERSION}'"  -o dist/litestream ./cmd/litestream
+	gon etc/gon.hcl
+	mv dist/litestream.zip dist/litestream-${LITESTREAM_VERSION}-darwin-arm64.zip
+	openssl dgst -sha256 dist/litestream-${LITESTREAM_VERSION}-darwin-arm64.zip
+
 clean:
 	rm -rf dist
 

README.md

@@ -6,7 +6,7 @@ Litestream
 ![test](https://github.com/benbjohnson/litestream/workflows/test/badge.svg)
 ==========
 
-Litestream is a standalone streaming replication tool for SQLite. It runs as a
+Litestream is a standalone disaster recovery tool for SQLite. It runs as a
 background process and safely replicates changes incrementally to another file
 or S3. Litestream only communicates with SQLite through the SQLite API so it
 will not corrupt your database.
@@ -33,35 +33,28 @@ energy into the project to help make it better:
 
 - Thanks to [Cory LaNou](https://twitter.com/corylanou) for giving early feedback and testing when Litestream was still pre-release.
 - Thanks to [Michael Lynch](https://github.com/mtlynch) for digging into issues and contributing to the documentation.
-- Thanks to [Kurt Mackey](https://twitter.com/mrkurt) for feedback and testing. Also, thanks to fly.io for providing testing resources.
+- Thanks to [Kurt Mackey](https://twitter.com/mrkurt) for feedback and testing.
 - Thanks to [Sam Weston](https://twitter.com/cablespaghetti) for figuring out how to run Litestream on Kubernetes and writing up the docs for it.
 - Thanks to [Rafael](https://github.com/netstx) & [Jungle Boogie](https://github.com/jungle-boogie) for helping to get OpenBSD release builds working.
-- Thanks to [Simon Gottschlag](https://github.com/simongottschlag), [Marin](https://github.com/supermarin),[Victor Björklund](https://github.com/victorbjorklund), [Jonathan Beri](https://twitter.com/beriberikix) [Yuri](https://github.com/yurivish), [Nathan Probst](https://github.com/nprbst), [Yann Coleuu](https://github.com/yanc0), and [Nicholas Grilly](https://twitter.com/ngrilly) for frequent feedback, testing, & support.
+- Thanks to [Simon Gottschlag](https://github.com/simongottschlag), [Marin](https://github.com/supermarin),[Victor Björklund](https://github.com/victorbjorklund), [Jonathan Beri](https://twitter.com/beriberikix) [Yuri](https://github.com/yurivish), [Nathan Probst](https://github.com/nprbst), [Yann Coleu](https://github.com/yanc0), and [Nicholas Grilly](https://twitter.com/ngrilly) for frequent feedback, testing, & support.
+
+Huge thanks to fly.io for their support and for contributing credits for testing and development!
 
 
+## Contribution Policy
 
+Initially, Litestream was closed to outside contributions. The goal was to
+reduce burnout by limiting the maintenance overhead of reviewing and validating
+third-party code. However, this policy is overly broad and has prevented small,
+easily testable patches from being contributed.
 
-## Open-source, not open-contribution
+Litestream is now open to code contributions for bug fixes only. Features carry
+a long-term maintenance burden so they will not be accepted at this time.
+Please [submit an issue][new-issue] if you have a feature you'd like to
+request.
 
-[Similar to SQLite](https://www.sqlite.org/copyright.html), Litestream is open
+If you find mistakes in the documentation, please submit a fix to the
-source but closed to code contributions. This keeps the code base free of
+[documentation repository][docs].
-proprietary or licensed code but it also helps me continue to maintain and build
-Litestream.
 
-As the author of [BoltDB](https://github.com/boltdb/bolt), I found that
+[new-issue]: https://github.com/benbjohnson/litestream/issues/new
-accepting and maintaining third party patches contributed to my burn out and
-I eventually archived the project. Writing databases & low-level replication
-tools involves nuance and simple one line changes can have profound and
-unexpected changes in correctness and performance. Small contributions
-typically required hours of my time to properly test and validate them.
-
-I am grateful for community involvement, bug reports, & feature requests. I do
-not wish to come off as anything but welcoming, however, I've
-made the decision to keep this project closed to contributions for my own
-mental health and long term viability of the project.
-
-The [documentation repository][docs] is MIT licensed and pull requests are welcome there.
-
-[releases]: https://github.com/benbjohnson/litestream/releases
 [docs]: https://github.com/benbjohnson/litestream.io
-

cmd/litestream/generations.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
 	"os"
+	"sort"
 	"text/tabwriter"
 	"time"
 
@@ -87,15 +87,17 @@ func (c *GenerationsCommand) Run(ctx context.Context, args []string) (err error)
 	for _, r := range replicas {
 		generations, err := r.Client.Generations(ctx)
 		if err != nil {
-			log.Printf("%s: cannot list generations: %s", r.Name(), err)
+			r.Logger().Error("cannot list generations", "error", err)
 			continue
 		}
 
+		sort.Strings(generations)
+
 		// Iterate over each generation for the replica.
 		for _, generation := range generations {
 			createdAt, updatedAt, err := r.GenerationTimeBounds(ctx, generation)
 			if err != nil {
-				log.Printf("%s: cannot determine generation time bounds: %s", r.Name(), err)
+				r.Logger().Error("cannot determine generation time bounds", "error", err)
 				continue
 			}
 

cmd/litestream/main.go

@@ -5,8 +5,7 @@ import (
 	"errors"
 	"flag"
 	"fmt"
-	"io/ioutil"
+	"log/slog"
-	"log"
 	"net/url"
 	"os"
 	"os/user"
@@ -17,6 +16,7 @@ import (
 	"strings"
 	"time"
 
+	"filippo.io/age"
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/abs"
 	"github.com/benbjohnson/litestream/file"
@@ -36,13 +36,11 @@ var (
 var errStop = errors.New("stop")
 
 func main() {
-	log.SetFlags(0)
-
 	m := NewMain()
 	if err := m.Run(context.Background(), os.Args[1:]); err == flag.ErrHelp || err == errStop {
 		os.Exit(1)
 	} else if err != nil {
-		log.Println(err)
+		slog.Error("failed to run", "error", err)
 		os.Exit(1)
 	}
 }
@@ -85,29 +83,26 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 		}
 
 		// Setup signal handler.
-		ctx, cancel := context.WithCancel(ctx)
 		signalCh := signalChan()
 
-		if err := c.Run(ctx); err != nil {
+		if err := c.Run(); err != nil {
 			return err
 		}
 
 		// Wait for signal to stop program.
 		select {
 		case err = <-c.execCh:
-			cancel()
+			slog.Info("subprocess exited, litestream shutting down")
-			fmt.Println("subprocess exited, litestream shutting down")
 		case sig := <-signalCh:
-			cancel()
+			slog.Info("signal received, litestream shutting down")
-			fmt.Println("signal received, litestream shutting down")
 
 			if c.cmd != nil {
-				fmt.Println("sending signal to exec process")
+				slog.Info("sending signal to exec process")
 				if err := c.cmd.Process.Signal(sig); err != nil {
 					return fmt.Errorf("cannot signal exec process: %w", err)
 				}
 
-				fmt.Println("waiting for exec process to close")
+				slog.Info("waiting for exec process to close")
 				if err := <-c.execCh; err != nil && !strings.HasPrefix(err.Error(), "signal:") {
 					return fmt.Errorf("cannot wait for exec process: %w", err)
 				}
@@ -118,7 +113,7 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 		if e := c.Close(); e != nil && err == nil {
 			err = e
 		}
-		fmt.Println("litestream shut down")
+		slog.Info("litestream shut down")
 		return err
 
 	case "restore":
@@ -174,6 +169,16 @@ type Config struct {
 	// Global S3 settings
 	AccessKeyID     string `yaml:"access-key-id"`
 	SecretAccessKey string `yaml:"secret-access-key"`
+
+	// Logging
+	Logging LoggingConfig `yaml:"logging"`
+}
+
+// LoggingConfig configures logging.
+type LoggingConfig struct {
+	Level  string `yaml:"level"`
+	Type   string `yaml:"type"`
+	Stderr bool   `yaml:"stderr"`
 }
 
 // propagateGlobalSettings copies global S3 settings to replica configs.
@@ -217,7 +222,7 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 	}
 
 	// Read configuration.
-	buf, err := ioutil.ReadFile(filename)
+	buf, err := os.ReadFile(filename)
 	if os.IsNotExist(err) {
 		return config, fmt.Errorf("config file not found: %s", filename)
 	} else if err != nil {
@@ -243,12 +248,43 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 	// Propage settings from global config to replica configs.
 	config.propagateGlobalSettings()
 
+	// Configure logging.
+	logOutput := os.Stdout
+	if config.Logging.Stderr {
+		logOutput = os.Stderr
+	}
+
+	logOptions := slog.HandlerOptions{
+		Level: slog.LevelInfo,
+	}
+
+	switch strings.ToUpper(config.Logging.Level) {
+	case "DEBUG":
+		logOptions.Level = slog.LevelDebug
+	case "WARN", "WARNING":
+		logOptions.Level = slog.LevelWarn
+	case "ERROR":
+		logOptions.Level = slog.LevelError
+	}
+
+	var logHandler slog.Handler
+	switch config.Logging.Type {
+	case "json":
+		logHandler = slog.NewJSONHandler(logOutput, &logOptions)
+	case "text", "":
+		logHandler = slog.NewTextHandler(logOutput, &logOptions)
+	}
+
+	// Set global default logger.
+	slog.SetDefault(slog.New(logHandler))
+
 	return config, nil
 }
 
 // DBConfig represents the configuration for a single database.
 type DBConfig struct {
 	Path               string         `yaml:"path"`
+	MetaPath           *string        `yaml:"meta-path"`
 	MonitorInterval    *time.Duration `yaml:"monitor-interval"`
 	CheckpointInterval *time.Duration `yaml:"checkpoint-interval"`
 	MinCheckpointPageN *int           `yaml:"min-checkpoint-page-count"`
@@ -268,6 +304,9 @@ func NewDBFromConfig(dbc *DBConfig) (*litestream.DB, error) {
 	db := litestream.NewDB(path)
 
 	// Override default database settings if specified in configuration.
+	if dbc.MetaPath != nil {
+		db.SetMetaPath(*dbc.MetaPath)
+	}
 	if dbc.MonitorInterval != nil {
 		db.MonitorInterval = *dbc.MonitorInterval
 	}
@@ -323,6 +362,12 @@ type ReplicaConfig struct {
 	User     string `yaml:"user"`
 	Password string `yaml:"password"`
 	KeyPath  string `yaml:"key-path"`
+
+	// Encryption identities and recipients
+	Age struct {
+		Identities []string `yaml:"identities"`
+		Recipients []string `yaml:"recipients"`
+	} `yaml:"age"`
 }
 
 // NewReplicaFromConfig instantiates a replica for a DB based on a config.
@@ -349,6 +394,22 @@ func NewReplicaFromConfig(c *ReplicaConfig, db *litestream.DB) (_ *litestream.Re
 	if v := c.ValidationInterval; v != nil {
 		r.ValidationInterval = *v
 	}
+	for _, str := range c.Age.Identities {
+		identities, err := age.ParseIdentities(strings.NewReader(str))
+		if err != nil {
+			return nil, err
+		}
+
+		r.AgeIdentities = append(r.AgeIdentities, identities...)
+	}
+	for _, str := range c.Age.Recipients {
+		recipients, err := age.ParseRecipients(strings.NewReader(str))
+		if err != nil {
+			return nil, err
+		}
+
+		r.AgeRecipients = append(r.AgeRecipients, recipients...)
+	}
 
 	// Build and set client on replica.
 	switch c.ReplicaType() {

cmd/litestream/main_notwindows.go

@@ -1,4 +1,4 @@
-// +build !windows
+//go:build !windows
 
 package main
 

cmd/litestream/main_test.go

@@ -1,7 +1,6 @@
 package main_test
 
 import (
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"testing"
@@ -16,7 +15,7 @@ func TestReadConfigFile(t *testing.T) {
 	// Ensure global AWS settings are propagated down to replica configurations.
 	t.Run("PropagateGlobalSettings", func(t *testing.T) {
 		filename := filepath.Join(t.TempDir(), "litestream.yml")
-		if err := ioutil.WriteFile(filename, []byte(`
+		if err := os.WriteFile(filename, []byte(`
 access-key-id: XXX
 secret-access-key: YYY
 
@@ -48,7 +47,7 @@ dbs:
 		os.Setenv("LITESTREAM_TEST_1872363", "s3://foo/bar")
 
 		filename := filepath.Join(t.TempDir(), "litestream.yml")
-		if err := ioutil.WriteFile(filename, []byte(`
+		if err := os.WriteFile(filename, []byte(`
 dbs:
   - path: $LITESTREAM_TEST_0129380
     replicas:
@@ -75,7 +74,7 @@ dbs:
 		os.Setenv("LITESTREAM_TEST_9847533", "s3://foo/bar")
 
 		filename := filepath.Join(t.TempDir(), "litestream.yml")
-		if err := ioutil.WriteFile(filename, []byte(`
+		if err := os.WriteFile(filename, []byte(`
 dbs:
   - path: /path/to/db
     replicas:

cmd/litestream/main_windows.go

@@ -1,11 +1,11 @@
-// +build windows
+//go:build windows
 
 package main
 
 import (
 	"context"
 	"io"
-	"log"
+	"log/slog"
 	"os"
 	"os/signal"
 
@@ -36,16 +36,16 @@ func runWindowsService(ctx context.Context) error {
 	defer elog.Close()
 
 	// Set eventlog as log writer while running.
-	log.SetOutput((*eventlogWriter)(elog))
+	slog.SetDefault(slog.New(slog.NewTextHandler((*eventlogWriter)(elog), nil)))
-	defer log.SetOutput(os.Stderr)
+	defer slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, nil)))
 
-	log.Print("Litestream service starting")
+	slog.Info("Litestream service starting")
 
 	if err := svc.Run(serviceName, &windowsService{ctx: ctx}); err != nil {
 		return errStop
 	}
 
-	log.Print("Litestream service stopped")
+	slog.Info("Litestream service stopped")
 	return nil
 }
 
@@ -63,13 +63,13 @@ func (s *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, stat
 	// Instantiate replication command and load configuration.
 	c := NewReplicateCommand()
 	if c.Config, err = ReadConfigFile(DefaultConfigPath(), true); err != nil {
-		log.Printf("cannot load configuration: %s", err)
+		slog.Error("cannot load configuration", "error", err)
 		return true, 1
 	}
 
 	// Execute replication command.
-	if err := c.Run(s.ctx); err != nil {
+	if err := c.Run(); err != nil {
-		log.Printf("cannot replicate: %s", err)
+		slog.Error("cannot replicate", "error", err)
 		statusCh <- svc.Status{State: svc.StopPending}
 		return true, 2
 	}
@@ -88,7 +88,7 @@ func (s *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, stat
 			case svc.Interrogate:
 				statusCh <- req.CurrentStatus
 			default:
-				log.Printf("Litestream service received unexpected change request cmd: %d", req.Cmd)
+				slog.Error("Litestream service received unexpected change request", "cmd", req.Cmd)
 			}
 		}
 	}

cmd/litestream/replicate.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
+	"log/slog"
 	"net"
 	"net/http"
 	_ "net/http/pprof"
@@ -42,7 +42,6 @@ func NewReplicateCommand() *ReplicateCommand {
 func (c *ReplicateCommand) ParseFlags(ctx context.Context, args []string) (err error) {
 	fs := flag.NewFlagSet("litestream-replicate", flag.ContinueOnError)
 	execFlag := fs.String("exec", "", "execute subcommand")
-	tracePath := fs.String("trace", "", "trace path")
 	configPath, noExpandEnv := registerConfigFlag(fs)
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
@@ -80,27 +79,17 @@ func (c *ReplicateCommand) ParseFlags(ctx context.Context, args []string) (err e
 		c.Config.Exec = *execFlag
 	}
 
-	// Enable trace logging.
-	if *tracePath != "" {
-		f, err := os.Create(*tracePath)
-		if err != nil {
-			return err
-		}
-		defer f.Close()
-		litestream.Tracef = log.New(f, "", log.LstdFlags|log.Lmicroseconds|log.LUTC|log.Lshortfile).Printf
-	}
-
 	return nil
 }
 
 // Run loads all databases specified in the configuration.
-func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
+func (c *ReplicateCommand) Run() (err error) {
 	// Display version information.
-	log.Printf("litestream %s", Version)
+	slog.Info("litestream", "version", Version)
 
 	// Setup databases.
 	if len(c.Config.DBs) == 0 {
-		log.Println("no databases specified in configuration")
+		slog.Error("no databases specified in configuration")
 	}
 
 	for _, dbConfig := range c.Config.DBs {
@@ -118,21 +107,22 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 
 	// Notify user that initialization is done.
 	for _, db := range c.DBs {
-		log.Printf("initialized db: %s", db.Path())
+		slog.Info("initialized db", "path", db.Path())
 		for _, r := range db.Replicas {
+			slog := slog.With("name", r.Name(), "type", r.Client.Type(), "sync-interval", r.SyncInterval)
 			switch client := r.Client.(type) {
 			case *file.ReplicaClient:
-				log.Printf("replicating to: name=%q type=%q path=%q", r.Name(), client.Type(), client.Path())
+				slog.Info("replicating to", "path", client.Path())
 			case *s3.ReplicaClient:
-				log.Printf("replicating to: name=%q type=%q bucket=%q path=%q region=%q endpoint=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, client.Region, client.Endpoint, r.SyncInterval)
+				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "region", client.Region, "endpoint", client.Endpoint)
 			case *gcs.ReplicaClient:
-				log.Printf("replicating to: name=%q type=%q bucket=%q path=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, r.SyncInterval)
+				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path)
 			case *abs.ReplicaClient:
-				log.Printf("replicating to: name=%q type=%q bucket=%q path=%q endpoint=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, client.Endpoint, r.SyncInterval)
+				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "endpoint", client.Endpoint)
 			case *sftp.ReplicaClient:
-				log.Printf("replicating to: name=%q type=%q host=%q user=%q path=%q sync-interval=%s", r.Name(), client.Type(), client.Host, client.User, client.Path, r.SyncInterval)
+				slog.Info("replicating to", "host", client.Host, "user", client.User, "path", client.Path)
 			default:
-				log.Printf("replicating to: name=%q type=%q", r.Name(), client.Type())
+				slog.Info("replicating to")
 			}
 		}
 	}
@@ -146,11 +136,11 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 			hostport = net.JoinHostPort("localhost", port)
 		}
 
-		log.Printf("serving metrics on http://%s/metrics", hostport)
+		slog.Info("serving metrics on", "url", fmt.Sprintf("http://%s/metrics", hostport))
 		go func() {
 			http.Handle("/metrics", promhttp.Handler())
 			if err := http.ListenAndServe(c.Config.Addr, nil); err != nil {
-				log.Printf("cannot start metrics server: %s", err)
+				slog.Error("cannot start metrics server", "error", err)
 			}
 		}()
 	}
@@ -162,7 +152,7 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 			return fmt.Errorf("cannot parse exec command: %w", err)
 		}
 
-		c.cmd = exec.CommandContext(ctx, execArgs[0], execArgs[1:]...)
+		c.cmd = exec.Command(execArgs[0], execArgs[1:]...)
 		c.cmd.Env = os.Environ()
 		c.cmd.Stdout = os.Stdout
 		c.cmd.Stderr = os.Stderr
@@ -178,8 +168,8 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 // Close closes all open databases.
 func (c *ReplicateCommand) Close() (err error) {
 	for _, db := range c.DBs {
-		if e := db.SoftClose(); e != nil {
+		if e := db.Close(context.Background()); e != nil {
-			log.Printf("error closing db: path=%s err=%s", db.Path(), e)
+			db.Logger.Error("error closing db", "error", e)
 			if err == nil {
 				err = e
 			}
@@ -215,8 +205,5 @@ Arguments:
 	-no-expand-env
 	    Disables environment variable expansion in configuration file.
 
-	-trace PATH
-	    Write verbose trace logging to PATH.
-
 `[1:], DefaultConfigPath())
 }

cmd/litestream/restore.go

@@ -5,7 +5,7 @@ import (
 	"errors"
 	"flag"
 	"fmt"
-	"log"
+	"log/slog"
 	"os"
 	"strconv"
 	"time"
@@ -19,7 +19,6 @@ type RestoreCommand struct{}
 // Run executes the command.
 func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
 	opt := litestream.NewRestoreOptions()
-	opt.Verbose = true
 
 	fs := flag.NewFlagSet("litestream-restore", flag.ContinueOnError)
 	configPath, noExpandEnv := registerConfigFlag(fs)
@@ -31,7 +30,6 @@ func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
 	ifDBNotExists := fs.Bool("if-db-not-exists", false, "")
 	ifReplicaExists := fs.Bool("if-replica-exists", false, "")
 	timestampStr := fs.String("timestamp", "", "timestamp")
-	verbose := fs.Bool("v", false, "verbose output")
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -48,11 +46,6 @@ func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
 		}
 	}
 
-	// Instantiate logger if verbose output is enabled.
-	if *verbose {
-		opt.Logger = log.New(os.Stderr, "", log.LstdFlags|log.Lmicroseconds)
-	}
-
 	// Determine replica & generation to restore from.
 	var r *litestream.Replica
 	if isURL(fs.Arg(0)) {
@@ -60,7 +53,7 @@ func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
 			return fmt.Errorf("cannot specify a replica URL and the -config flag")
 		}
 		if r, err = c.loadFromURL(ctx, fs.Arg(0), *ifDBNotExists, &opt); err == errSkipDBExists {
-			fmt.Println("database already exists, skipping")
+			slog.Info("database already exists, skipping")
 			return nil
 		} else if err != nil {
 			return err
@@ -70,7 +63,7 @@ func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
 			*configPath = DefaultConfigPath()
 		}
 		if r, err = c.loadFromConfig(ctx, fs.Arg(0), *configPath, !*noExpandEnv, *ifDBNotExists, &opt); err == errSkipDBExists {
-			fmt.Println("database already exists, skipping")
+			slog.Info("database already exists, skipping")
 			return nil
 		} else if err != nil {
 			return err
@@ -81,7 +74,7 @@ func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
 	// If optional flag set, return success. Useful for automated recovery.
 	if opt.Generation == "" {
 		if *ifReplicaExists {
-			fmt.Println("no matching backups found")
+			slog.Info("no matching backups found")
 			return nil
 		}
 		return fmt.Errorf("no matching backups found")
@@ -204,9 +197,6 @@ Arguments:
 	    Determines the number of WAL files downloaded in parallel.
 	    Defaults to `+strconv.Itoa(litestream.DefaultRestoreParallelism)+`.
 
-	-v
-	    Verbose output.
-
 
 Examples:
 

cmd/litestream/snapshots.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
+	"log/slog"
 	"os"
 	"text/tabwriter"
 	"time"
@@ -82,7 +82,7 @@ func (c *SnapshotsCommand) Run(ctx context.Context, args []string) (err error) {
 	for _, r := range replicas {
 		infos, err := r.Snapshots(ctx)
 		if err != nil {
-			log.Printf("cannot determine snapshots: %s", err)
+			slog.Error("cannot determine snapshots", "error", err)
 			continue
 		}
 		for _, info := range infos {

cmd/litestream/wal.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
 	"os"
 	"text/tabwriter"
 	"time"
@@ -86,7 +85,7 @@ func (c *WALCommand) Run(ctx context.Context, args []string) (err error) {
 			generations = []string{*generation}
 		} else {
 			if generations, err = r.Client.Generations(ctx); err != nil {
-				log.Printf("%s: cannot determine generations: %s", r.Name(), err)
+				r.Logger().Error("cannot determine generations", "error", err)
 				continue
 			}
 		}
@@ -102,7 +101,7 @@ func (c *WALCommand) Run(ctx context.Context, args []string) (err error) {
 				for itr.Next() {
 					info := itr.WALSegment()
 
-					fmt.Fprintf(w, "%s\t%s\t%d\t%d\t%d\t%s\n",
+					fmt.Fprintf(w, "%s\t%s\t%x\t%d\t%d\t%s\n",
 						r.Name(),
 						info.Generation,
 						info.Index,
@@ -113,7 +112,7 @@ func (c *WALCommand) Run(ctx context.Context, args []string) (err error) {
 				}
 				return itr.Close()
 			}(); err != nil {
-				log.Printf("%s: cannot fetch wal segments: %s", r.Name(), err)
+				r.Logger().Error("cannot fetch wal segments", "error", err)
 				continue
 			}
 		}

db.go

@@ -10,8 +10,7 @@ import (
 	"fmt"
 	"hash/crc64"
 	"io"
-	"io/ioutil"
+	"log/slog"
-	"log"
 	"math"
 	"math/rand"
 	"os"
@@ -31,6 +30,7 @@ const (
 	DefaultCheckpointInterval = 1 * time.Minute
 	DefaultMinCheckpointPageN = 1000
 	DefaultMaxCheckpointPageN = 10000
+	DefaultTruncatePageN      = 500000
 )
 
 // MaxIndex is the maximum possible WAL index.
@@ -44,11 +44,13 @@ const BusyTimeout = 1 * time.Second
 type DB struct {
 	mu       sync.RWMutex
 	path     string        // part to database
+	metaPath string        // Path to the database metadata.
 	db       *sql.DB       // target database
 	f        *os.File      // long-running db file descriptor
 	rtx      *sql.Tx       // long running read transaction
 	pageSize int           // page size, in bytes
 	notify   chan struct{} // closes on WAL change
+	chkMu    sync.Mutex    // checkpoint lock
 
 	fileInfo os.FileInfo // db info cached during init
 	dirInfo  os.FileInfo // parent dir info cached during init
@@ -83,6 +85,16 @@ type DB struct {
 	// unbounded if there are always read transactions occurring.
 	MaxCheckpointPageN int
 
+	// Threshold of WAL size, in pages, before a forced truncation checkpoint.
+	// A forced truncation checkpoint will block new transactions and wait for
+	// existing transactions to finish before issuing a checkpoint and
+	// truncating the WAL.
+	//
+	// If zero, no truncates are forced. This can cause the WAL to grow
+	// unbounded if there's a sudden spike of changes between other
+	// checkpoints.
+	TruncatePageN int
+
 	// Time between automatic checkpoints in the WAL. This is done to allow
 	// more fine-grained WAL files so that restores can be performed with
 	// better precision.
@@ -94,18 +106,60 @@ type DB struct {
 	// List of replicas for the database.
 	// Must be set before calling Open().
 	Replicas []*Replica
+
+	// Where to send log messages, defaults to global slog with databas epath.
+	Logger *slog.Logger
 }
 
 // NewDB returns a new instance of DB for a given path.
 func NewDB(path string) *DB {
+	dir, file := filepath.Split(path)
+
 	db := &DB{
 		path:     path,
+		metaPath: filepath.Join(dir, "."+file+MetaDirSuffix),
 		notify:   make(chan struct{}),
 
 		MinCheckpointPageN: DefaultMinCheckpointPageN,
 		MaxCheckpointPageN: DefaultMaxCheckpointPageN,
+		TruncatePageN:      DefaultTruncatePageN,
 		CheckpointInterval: DefaultCheckpointInterval,
 		MonitorInterval:    DefaultMonitorInterval,
+		Logger:             slog.With("db", path),
+	}
+
+	db.dbSizeGauge = dbSizeGaugeVec.WithLabelValues(db.path)
+	db.walSizeGauge = walSizeGaugeVec.WithLabelValues(db.path)
+	db.totalWALBytesCounter = totalWALBytesCounterVec.WithLabelValues(db.path)
+	db.shadowWALIndexGauge = shadowWALIndexGaugeVec.WithLabelValues(db.path)
+	db.shadowWALSizeGauge = shadowWALSizeGaugeVec.WithLabelValues(db.path)
+	db.syncNCounter = syncNCounterVec.WithLabelValues(db.path)
+	db.syncErrorNCounter = syncErrorNCounterVec.WithLabelValues(db.path)
+	db.syncSecondsCounter = syncSecondsCounterVec.WithLabelValues(db.path)
+	db.checkpointNCounterVec = checkpointNCounterVec.MustCurryWith(prometheus.Labels{"db": db.path})
+	db.checkpointErrorNCounterVec = checkpointErrorNCounterVec.MustCurryWith(prometheus.Labels{"db": db.path})
+	db.checkpointSecondsCounterVec = checkpointSecondsCounterVec.MustCurryWith(prometheus.Labels{"db": db.path})
+
+	db.ctx, db.cancel = context.WithCancel(context.Background())
+
+	return db
+}
+
+// NewDB returns a new instance of DB for a given path.
+func NewLSDB(path string, metapath string) *DB {
+	_, file := filepath.Split(path)
+
+	db := &DB{
+		path:     path,
+		metaPath: filepath.Join(metapath, "."+file+MetaDirSuffix),
+		notify:   make(chan struct{}),
+
+		MinCheckpointPageN: DefaultMinCheckpointPageN,
+		MaxCheckpointPageN: DefaultMaxCheckpointPageN,
+		TruncatePageN:      DefaultTruncatePageN,
+		CheckpointInterval: DefaultCheckpointInterval,
+		MonitorInterval:    DefaultMonitorInterval,
+		Logger:             slog.With("db", path),
 	}
 
 	db.dbSizeGauge = dbSizeGaugeVec.WithLabelValues(db.path)
@@ -142,20 +196,24 @@ func (db *DB) WALPath() string {
 
 // MetaPath returns the path to the database metadata.
 func (db *DB) MetaPath() string {
-	dir, file := filepath.Split(db.path)
+	return db.metaPath
-	return filepath.Join(dir, "."+file+MetaDirSuffix)
+}
+
+// SetMetaPath sets the path to database metadata.
+func (db *DB) SetMetaPath(mp string) {
+	db.metaPath = mp
 }
 
 // GenerationNamePath returns the path of the name of the current generation.
 func (db *DB) GenerationNamePath() string {
-	return filepath.Join(db.MetaPath(), "generation")
+	return filepath.Join(db.metaPath, "generation")
 }
 
 // GenerationPath returns the path of a single generation.
 // Panics if generation is blank.
 func (db *DB) GenerationPath(generation string) string {
 	assert(generation != "", "generation name required")
-	return filepath.Join(db.MetaPath(), "generations", generation)
+	return filepath.Join(db.metaPath, "generations", generation)
 }
 
 // ShadowWALDir returns the path of the shadow wal directory.
@@ -182,7 +240,7 @@ func (db *DB) CurrentShadowWALPath(generation string) (string, error) {
 
 // CurrentShadowWALIndex returns the current WAL index & total size.
 func (db *DB) CurrentShadowWALIndex(generation string) (index int, size int64, err error) {
-	fis, err := ioutil.ReadDir(filepath.Join(db.GenerationPath(generation), "wal"))
+	des, err := os.ReadDir(filepath.Join(db.GenerationPath(generation), "wal"))
 	if os.IsNotExist(err) {
 		return 0, 0, nil // no wal files written for generation
 	} else if err != nil {
@@ -190,7 +248,14 @@ func (db *DB) CurrentShadowWALIndex(generation string) (index int, size int64, e
 	}
 
 	// Find highest wal index.
-	for _, fi := range fis {
+	for _, de := range des {
+		fi, err := de.Info()
+		if os.IsNotExist(err) {
+			continue // file was deleted after os.ReadDir returned
+		} else if err != nil {
+			return 0, 0, err
+		}
+
 		if v, err := ParseWALPath(fi.Name()); err != nil {
 			continue // invalid wal filename
 		} else if v > index {
@@ -278,7 +343,7 @@ func (db *DB) Open() (err error) {
 	}
 
 	// Clear old temporary files that my have been left from a crash.
-	if err := removeTmpFiles(db.MetaPath()); err != nil {
+	if err := removeTmpFiles(db.metaPath); err != nil {
 		return fmt.Errorf("cannot remove tmp files: %w", err)
 	}
 
@@ -291,26 +356,12 @@ func (db *DB) Open() (err error) {
 	return nil
 }
 
-// Close releases the read lock & closes the database. This method should only
+// Close flushes outstanding WAL writes to replicas, releases the read lock,
-// be called by tests as it causes the underlying database to be checkpointed.
+// and closes the database. Takes a context for final sync.
-func (db *DB) Close() (err error) {
+func (db *DB) Close(ctx context.Context) (err error) {
-	return db.close(false)
-}
-
-// SoftClose closes everything but the underlying db connection. This method
-// is available because the binary needs to avoid closing the database on exit
-// to prevent autocheckpointing.
-func (db *DB) SoftClose() (err error) {
-	return db.close(true)
-}
-
-func (db *DB) close(soft bool) (err error) {
 	db.cancel()
 	db.wg.Wait()
 
-	// Start a new context for shutdown since we canceled the DB context.
-	ctx := context.Background()
-
 	// Perform a final db sync, if initialized.
 	if db.db != nil {
 		if e := db.Sync(ctx); e != nil && err == nil {
@@ -325,7 +376,7 @@ func (db *DB) close(soft bool) (err error) {
 				err = e
 			}
 		}
-		r.Stop(!soft)
+		r.Stop(true)
 	}
 
 	// Release the read lock to allow other applications to handle checkpointing.
@@ -335,14 +386,18 @@ func (db *DB) close(soft bool) (err error) {
 		}
 	}
 
-	// Only perform full close if this is not a soft close.
+	if db.db != nil {
-	// This closes the underlying database connection which can clean up the WAL.
-	if !soft && db.db != nil {
 		if e := db.db.Close(); e != nil && err == nil {
 			err = e
 		}
 	}
 
+	if db.f != nil {
+		if e := db.f.Close(); e != nil && err == nil {
+			err = e
+		}
+	}
+
 	return err
 }
 
@@ -392,8 +447,9 @@ func (db *DB) init() (err error) {
 	dsn := db.path
 	dsn += fmt.Sprintf("?_busy_timeout=%d", BusyTimeout.Milliseconds())
 
-	// Connect to SQLite database.
+	// Connect to SQLite database. Use the driver registered with a hook to
-	if db.db, err = sql.Open("sqlite3", dsn); err != nil {
+	// prevent WAL files from being removed.
+	if db.db, err = sql.Open("litestream-sqlite3", dsn); err != nil {
 		return err
 	}
 
@@ -453,13 +509,13 @@ func (db *DB) init() (err error) {
 	}
 
 	// Ensure meta directory structure exists.
-	if err := internal.MkdirAll(db.MetaPath(), db.dirInfo); err != nil {
+	if err := internal.MkdirAll(db.metaPath, db.dirInfo); err != nil {
 		return err
 	}
 
 	// If we have an existing shadow WAL, ensure the headers match.
 	if err := db.verifyHeadersMatch(); err != nil {
-		log.Printf("%s: init: cannot determine last wal position, clearing generation; %s", db.path, err)
+		db.Logger.Warn("init: cannot determine last wal position, clearing generation", "error", err)
 		if err := os.Remove(db.GenerationNamePath()); err != nil && !os.IsNotExist(err) {
 			return fmt.Errorf("remove generation name: %w", err)
 		}
@@ -529,8 +585,8 @@ func (db *DB) cleanGenerations() error {
 		return err
 	}
 
-	dir := filepath.Join(db.MetaPath(), "generations")
+	dir := filepath.Join(db.metaPath, "generations")
-	fis, err := ioutil.ReadDir(dir)
+	fis, err := os.ReadDir(dir)
 	if os.IsNotExist(err) {
 		return nil
 	} else if err != nil {
@@ -577,7 +633,7 @@ func (db *DB) cleanWAL() error {
 
 	// Remove all WAL files for the generation before the lowest index.
 	dir := db.ShadowWALDir(generation)
-	fis, err := ioutil.ReadDir(dir)
+	fis, err := os.ReadDir(dir)
 	if os.IsNotExist(err) {
 		return nil
 	} else if err != nil {
@@ -633,7 +689,7 @@ func (db *DB) releaseReadLock() error {
 // CurrentGeneration returns the name of the generation saved to the "generation"
 // file in the meta data directory. Returns empty string if none exists.
 func (db *DB) CurrentGeneration() (string, error) {
-	buf, err := ioutil.ReadFile(db.GenerationNamePath())
+	buf, err := os.ReadFile(db.GenerationNamePath())
 	if os.IsNotExist(err) {
 		return "", nil
 	} else if err != nil {
@@ -659,7 +715,7 @@ func (db *DB) createGeneration() (string, error) {
 	generation := hex.EncodeToString(buf)
 
 	// Generate new directory.
-	dir := filepath.Join(db.MetaPath(), "generations", generation)
+	dir := filepath.Join(db.metaPath, "generations", generation)
 	if err := internal.MkdirAll(dir, db.dirInfo); err != nil {
 		return "", err
 	}
@@ -675,7 +731,7 @@ func (db *DB) createGeneration() (string, error) {
 	if db.fileInfo != nil {
 		mode = db.fileInfo.Mode()
 	}
-	if err := ioutil.WriteFile(generationNamePath+".tmp", []byte(generation+"\n"), mode); err != nil {
+	if err := os.WriteFile(generationNamePath+".tmp", []byte(generation+"\n"), mode); err != nil {
 		return "", fmt.Errorf("write generation temp file: %w", err)
 	}
 	uid, gid := internal.Fileinfo(db.fileInfo)
@@ -701,7 +757,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 	if err := db.init(); err != nil {
 		return err
 	} else if db.db == nil {
-		Tracef("%s: sync: no database found", db.path)
+		db.Logger.Debug("sync: no database found")
 		return nil
 	}
 
@@ -727,7 +783,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 	if err != nil {
 		return fmt.Errorf("cannot verify wal state: %w", err)
 	}
-	Tracef("%s: sync: info=%#v", db.path, info)
+	db.Logger.Debug("sync", "info", &info)
 
 	// Track if anything in the shadow WAL changes and then notify at the end.
 	changed := info.walSize != info.shadowWALSize || info.restart || info.reason != ""
@@ -738,7 +794,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 		if info.generation, err = db.createGeneration(); err != nil {
 			return fmt.Errorf("create generation: %w", err)
 		}
-		log.Printf("%s: sync: new generation %q, %s", db.path, info.generation, info.reason)
+		db.Logger.Info("sync: new generation", "generation", info.generation, "reason", info.reason)
 
 		// Clear shadow wal info.
 		info.shadowWALPath = db.ShadowWALPath(info.generation, 0)
@@ -749,7 +805,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 	}
 
 	// Synchronize real WAL with current shadow WAL.
-	newWALSize, err := db.syncWAL(info)
+	origWALSize, newWALSize, err := db.syncWAL(info)
 	if err != nil {
 		return fmt.Errorf("sync wal: %w", err)
 	}
@@ -758,7 +814,9 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 	// If WAL size is greater than min threshold, attempt checkpoint.
 	var checkpoint bool
 	checkpointMode := CheckpointModePassive
-	if db.MaxCheckpointPageN > 0 && newWALSize >= calcWALSize(db.pageSize, db.MaxCheckpointPageN) {
+	if db.TruncatePageN > 0 && origWALSize >= calcWALSize(db.pageSize, db.TruncatePageN) {
+		checkpoint, checkpointMode = true, CheckpointModeTruncate
+	} else if db.MaxCheckpointPageN > 0 && newWALSize >= calcWALSize(db.pageSize, db.MaxCheckpointPageN) {
 		checkpoint, checkpointMode = true, CheckpointModeRestart
 	} else if newWALSize >= calcWALSize(db.pageSize, db.MinCheckpointPageN) {
 		checkpoint = true
@@ -792,7 +850,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 		db.notify = make(chan struct{})
 	}
 
-	Tracef("%s: sync: ok", db.path)
+	db.Logger.Debug("sync: ok")
 
 	return nil
 }
@@ -917,29 +975,29 @@ type syncInfo struct {
 }
 
 // syncWAL copies pending bytes from the real WAL to the shadow WAL.
-func (db *DB) syncWAL(info syncInfo) (newSize int64, err error) {
+func (db *DB) syncWAL(info syncInfo) (origSize int64, newSize int64, err error) {
 	// Copy WAL starting from end of shadow WAL. Exit if no new shadow WAL needed.
-	newSize, err = db.copyToShadowWAL(info.shadowWALPath)
+	origSize, newSize, err = db.copyToShadowWAL(info.shadowWALPath)
 	if err != nil {
-		return newSize, fmt.Errorf("cannot copy to shadow wal: %w", err)
+		return origSize, newSize, fmt.Errorf("cannot copy to shadow wal: %w", err)
 	} else if !info.restart {
-		return newSize, nil // If no restart required, exit.
+		return origSize, newSize, nil // If no restart required, exit.
 	}
 
 	// Parse index of current shadow WAL file.
 	dir, base := filepath.Split(info.shadowWALPath)
 	index, err := ParseWALPath(base)
 	if err != nil {
-		return 0, fmt.Errorf("cannot parse shadow wal filename: %s", base)
+		return 0, 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))
 	newSize, err = db.initShadowWALFile(newShadowWALPath)
 	if err != nil {
-		return 0, fmt.Errorf("cannot init shadow wal file: name=%s err=%w", newShadowWALPath, err)
+		return 0, 0, fmt.Errorf("cannot init shadow wal file: name=%s err=%w", newShadowWALPath, err)
 	}
-	return newSize, nil
+	return origSize, newSize, nil
 }
 
 func (db *DB) initShadowWALFile(filename string) (int64, error) {
@@ -968,87 +1026,103 @@ func (db *DB) initShadowWALFile(filename string) (int64, error) {
 	}
 	if err := internal.MkdirAll(filepath.Dir(filename), db.dirInfo); err != nil {
 		return 0, err
-	} else if err := ioutil.WriteFile(filename, hdr, mode); err != nil {
+	} else if err := os.WriteFile(filename, hdr, mode); err != nil {
 		return 0, err
 	}
 	uid, gid := internal.Fileinfo(db.fileInfo)
 	_ = os.Chown(filename, uid, gid)
 
 	// Copy as much shadow WAL as available.
-	newSize, err := db.copyToShadowWAL(filename)
+	_, newSize, err := db.copyToShadowWAL(filename)
 	if err != nil {
 		return 0, fmt.Errorf("cannot copy to new shadow wal: %w", err)
 	}
 	return newSize, nil
 }
 
-func (db *DB) copyToShadowWAL(filename string) (newSize int64, err error) {
+func (db *DB) copyToShadowWAL(filename string) (origWalSize int64, newSize int64, err error) {
-	Tracef("%s: copy-shadow: %s", db.path, filename)
+	logger := db.Logger.With("filename", filename)
+	logger.Debug("copy-shadow")
 
 	r, err := os.Open(db.WALPath())
 	if err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 	defer r.Close()
 
+	fi, err := r.Stat()
+	if err != nil {
+		return 0, 0, err
+	}
+	origWalSize = frameAlign(fi.Size(), db.pageSize)
+
 	w, err := os.OpenFile(filename, os.O_RDWR, 0666)
 	if err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 	defer w.Close()
 
-	fi, err := w.Stat()
+	fi, err = w.Stat()
 	if err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 	origSize := frameAlign(fi.Size(), db.pageSize)
 
 	// Read shadow WAL header to determine byte order for checksum & salt.
 	hdr := make([]byte, WALHeaderSize)
 	if _, err := io.ReadFull(w, hdr); err != nil {
-		return 0, fmt.Errorf("read header: %w", err)
+		return 0, 0, fmt.Errorf("read header: %w", err)
 	}
 	hsalt0 := binary.BigEndian.Uint32(hdr[16:])
 	hsalt1 := binary.BigEndian.Uint32(hdr[20:])
 
 	bo, err := headerByteOrder(hdr)
 	if err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 
 	// Read previous checksum.
 	chksum0, chksum1, err := readLastChecksumFrom(w, db.pageSize)
 	if err != nil {
-		return 0, fmt.Errorf("last checksum: %w", err)
+		return 0, 0, fmt.Errorf("last checksum: %w", err)
 	}
 
+	// Write to a temporary shadow file.
+	tempFilename := filename + ".tmp"
+	defer os.Remove(tempFilename)
+
+	f, err := internal.CreateFile(tempFilename, db.fileInfo)
+	if err != nil {
+		return 0, 0, fmt.Errorf("create temp file: %w", err)
+	}
+	defer f.Close()
+
 	// Seek to correct position on real wal.
 	if _, err := r.Seek(origSize, io.SeekStart); err != nil {
-		return 0, fmt.Errorf("real wal seek: %w", err)
+		return 0, 0, fmt.Errorf("real wal seek: %w", err)
 	} else if _, err := w.Seek(origSize, io.SeekStart); err != nil {
-		return 0, fmt.Errorf("shadow wal seek: %w", err)
+		return 0, 0, fmt.Errorf("shadow wal seek: %w", err)
 	}
 
 	// Read through WAL from last position to find the page of the last
 	// committed transaction.
 	frame := make([]byte, db.pageSize+WALFrameHeaderSize)
-	var buf bytes.Buffer
 	offset := origSize
 	lastCommitSize := origSize
 	for {
 		// Read next page from WAL file.
 		if _, err := io.ReadFull(r, frame); err == io.EOF || err == io.ErrUnexpectedEOF {
-			Tracef("%s: copy-shadow: break %s @ %d; err=%s", db.path, filename, offset, err)
+			logger.Debug("copy-shadow: break", "offset", offset, "error", err)
 			break // end of file or partial page
 		} else if err != nil {
-			return 0, fmt.Errorf("read wal: %w", err)
+			return 0, 0, fmt.Errorf("read wal: %w", err)
 		}
 
 		// Read frame salt & compare to header salt. Stop reading on mismatch.
 		salt0 := binary.BigEndian.Uint32(frame[8:])
 		salt1 := binary.BigEndian.Uint32(frame[12:])
 		if salt0 != hsalt0 || salt1 != hsalt1 {
-			Tracef("%s: copy-shadow: break: salt mismatch", db.path)
+			logger.Debug("copy-shadow: break: salt mismatch")
 			break
 		}
 
@@ -1058,38 +1132,60 @@ func (db *DB) copyToShadowWAL(filename string) (newSize int64, err error) {
 		chksum0, chksum1 = Checksum(bo, chksum0, chksum1, frame[:8])  // frame header
 		chksum0, chksum1 = Checksum(bo, chksum0, chksum1, frame[24:]) // frame data
 		if chksum0 != fchksum0 || chksum1 != fchksum1 {
-			Tracef("%s: copy shadow: checksum mismatch, skipping: offset=%d (%x,%x) != (%x,%x)", db.path, offset, chksum0, chksum1, fchksum0, fchksum1)
+			logger.Debug("copy shadow: checksum mismatch, skipping", "offset", offset, "check", fmt.Sprintf("(%x,%x) != (%x,%x)", chksum0, chksum1, fchksum0, fchksum1))
 			break
 		}
 
-		// Add page to the new size of the shadow WAL.
+		// Write page to temporary WAL file.
-		buf.Write(frame)
+		if _, err := f.Write(frame); err != nil {
+			return 0, 0, fmt.Errorf("write temp shadow wal: %w", err)
+		}
 
-		Tracef("%s: copy-shadow: ok %s offset=%d salt=%x %x", db.path, filename, offset, salt0, salt1)
+		logger.Debug("copy-shadow: ok", "offset", offset, "salt", fmt.Sprintf("%x %x", salt0, salt1))
 		offset += int64(len(frame))
 
-		// Flush to shadow WAL if commit record.
+		// Update new size if written frame was a commit record.
 		newDBSize := binary.BigEndian.Uint32(frame[4:])
 		if newDBSize != 0 {
-			if _, err := buf.WriteTo(w); err != nil {
-				return 0, fmt.Errorf("write shadow wal: %w", err)
-			}
-			buf.Reset()
 			lastCommitSize = offset
 		}
 	}
 
-	// Sync & close.
+	// If no WAL writes found, exit.
+	if origSize == lastCommitSize {
+		return origSize, lastCommitSize, nil
+	}
+
+	walByteN := lastCommitSize - origSize
+
+	// Move to beginning of temporary file.
+	if _, err := f.Seek(0, io.SeekStart); err != nil {
+		return 0, 0, fmt.Errorf("temp file seek: %w", err)
+	}
+
+	// Copy from temporary file to shadow WAL.
+	if _, err := io.Copy(w, &io.LimitedReader{R: f, N: walByteN}); err != nil {
+		return 0, 0, fmt.Errorf("write shadow file: %w", err)
+	}
+
+	// Close & remove temporary file.
+	if err := f.Close(); err != nil {
+		return 0, 0, err
+	} else if err := os.Remove(tempFilename); err != nil {
+		return 0, 0, err
+	}
+
+	// Sync & close shadow WAL.
 	if err := w.Sync(); err != nil {
-		return 0, err
+		return 0, 0, err
 	} else if err := w.Close(); err != nil {
-		return 0, err
+		return 0, 0, err
 	}
 
 	// Track total number of bytes written to WAL.
-	db.totalWALBytesCounter.Add(float64(lastCommitSize - origSize))
+	db.totalWALBytesCounter.Add(float64(walByteN))
 
-	return lastCommitSize, nil
+	return origWalSize, lastCommitSize, nil
 }
 
 // ShadowWALReader opens a reader for a shadow WAL file at a given position.
@@ -1243,9 +1339,15 @@ func (db *DB) Checkpoint(ctx context.Context, mode string) (err error) {
 	return db.checkpoint(ctx, generation, mode)
 }
 
-// checkpointAndInit performs a checkpoint on the WAL file and initializes a
+// checkpoint performs a checkpoint on the WAL file and initializes a
 // new shadow WAL file.
 func (db *DB) checkpoint(ctx context.Context, generation, mode string) error {
+	// Try getting a checkpoint lock, will fail during snapshots.
+	if !db.chkMu.TryLock() {
+		return nil
+	}
+	defer db.chkMu.Unlock()
+
 	shadowWALPath, err := db.CurrentShadowWALPath(generation)
 	if err != nil {
 		return err
@@ -1258,7 +1360,7 @@ func (db *DB) checkpoint(ctx context.Context, generation, mode string) error {
 	}
 
 	// Copy shadow WAL before checkpoint to copy as much as possible.
-	if _, err := db.copyToShadowWAL(shadowWALPath); err != nil {
+	if _, _, err := db.copyToShadowWAL(shadowWALPath); err != nil {
 		return fmt.Errorf("cannot copy to end of shadow wal before checkpoint: %w", err)
 	}
 
@@ -1293,7 +1395,7 @@ func (db *DB) checkpoint(ctx context.Context, generation, mode string) error {
 	}
 
 	// Copy the end of the previous WAL before starting a new shadow WAL.
-	if _, err := db.copyToShadowWAL(shadowWALPath); err != nil {
+	if _, _, err := db.copyToShadowWAL(shadowWALPath); err != nil {
 		return fmt.Errorf("cannot copy to end of shadow wal: %w", err)
 	}
 
@@ -1352,7 +1454,7 @@ func (db *DB) execCheckpoint(mode string) (err error) {
 	if err := db.db.QueryRow(rawsql).Scan(&row[0], &row[1], &row[2]); err != nil {
 		return err
 	}
-	Tracef("%s: checkpoint: mode=%v (%d,%d,%d)", db.path, mode, row[0], row[1], row[2])
+	db.Logger.Debug("checkpoint", "mode", mode, "result", fmt.Sprintf("%d,%d,%d", row[0], row[1], row[2]))
 
 	// Reacquire the read lock immediately after the checkpoint.
 	if err := db.acquireReadLock(); err != nil {
@@ -1377,7 +1479,7 @@ func (db *DB) monitor() {
 
 		// Sync the database to the shadow WAL.
 		if err := db.Sync(db.ctx); err != nil && !errors.Is(err, context.Canceled) {
-			log.Printf("%s: sync error: %s", db.path, err)
+			db.Logger.Error("sync error", "error", err)
 		}
 	}
 }
@@ -1481,6 +1583,19 @@ func (db *DB) CRC64(ctx context.Context) (uint64, Pos, error) {
 	return h.Sum64(), pos, nil
 }
 
+// BeginSnapshot takes an internal snapshot lock preventing checkpoints.
+//
+// When calling this the caller must also call EndSnapshot() once the snapshot
+// is finished.
+func (db *DB) BeginSnapshot() {
+	db.chkMu.Lock()
+}
+
+// EndSnapshot releases the internal snapshot lock that prevents checkpoints.
+func (db *DB) EndSnapshot() {
+	db.chkMu.Unlock()
+}
+
 // DefaultRestoreParallelism is the default parallelism when downloading WAL files.
 const DefaultRestoreParallelism = 8
 
@@ -1508,10 +1623,6 @@ type RestoreOptions struct {
 
 	// Specifies how many WAL files are downloaded in parallel during restore.
 	Parallelism int
-
-	// Logging settings.
-	Logger  *log.Logger
-	Verbose bool
 }
 
 // NewRestoreOptions returns a new instance of RestoreOptions with defaults.

db_test.go

@@ -3,7 +3,6 @@ package litestream_test
 import (
 	"context"
 	"database/sql"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strings"
@@ -264,14 +263,14 @@ func TestDB_Sync(t *testing.T) {
 		// Checkpoint & fully close which should close WAL file.
 		if err := db.Checkpoint(context.Background(), litestream.CheckpointModeTruncate); err != nil {
 			t.Fatal(err)
-		} else if err := db.Close(); err != nil {
+		} else if err := db.Close(context.Background()); err != nil {
 			t.Fatal(err)
 		} else if err := sqldb.Close(); err != nil {
 			t.Fatal(err)
 		}
 
-		// Verify WAL does not exist.
+		// Remove WAL file.
-		if _, err := os.Stat(db.WALPath()); !os.IsNotExist(err) {
+		if err := os.Remove(db.WALPath()); err != nil {
 			t.Fatal(err)
 		}
 
@@ -314,7 +313,7 @@ func TestDB_Sync(t *testing.T) {
 		}
 
 		// Fully close which should close WAL file.
-		if err := db.Close(); err != nil {
+		if err := db.Close(context.Background()); err != nil {
 			t.Fatal(err)
 		} else if err := sqldb.Close(); err != nil {
 			t.Fatal(err)
@@ -367,16 +366,16 @@ func TestDB_Sync(t *testing.T) {
 		pos0, err := db.Pos()
 		if err != nil {
 			t.Fatal(err)
-		} else if err := db.Close(); err != nil {
+		} else if err := db.Close(context.Background()); err != nil {
 			t.Fatal(err)
 		}
 
 		// Read existing file, update header checksum, and write back only header
 		// to simulate a header with a mismatched checksum.
 		shadowWALPath := db.ShadowWALPath(pos0.Generation, pos0.Index)
-		if buf, err := ioutil.ReadFile(shadowWALPath); err != nil {
+		if buf, err := os.ReadFile(shadowWALPath); err != nil {
 			t.Fatal(err)
-		} else if err := ioutil.WriteFile(shadowWALPath, append(buf[:litestream.WALHeaderSize-8], 0, 0, 0, 0, 0, 0, 0, 0), 0600); err != nil {
+		} else if err := os.WriteFile(shadowWALPath, append(buf[:litestream.WALHeaderSize-8], 0, 0, 0, 0, 0, 0, 0, 0), 0600); err != nil {
 			t.Fatal(err)
 		}
 
@@ -413,7 +412,7 @@ func TestDB_Sync(t *testing.T) {
 		}
 
 		// Close & truncate shadow WAL to simulate a partial header write.
-		if err := db.Close(); err != nil {
+		if err := db.Close(context.Background()); err != nil {
 			t.Fatal(err)
 		} else if err := os.Truncate(db.ShadowWALPath(pos0.Generation, pos0.Index), litestream.WALHeaderSize-1); err != nil {
 			t.Fatal(err)
@@ -458,7 +457,7 @@ func TestDB_Sync(t *testing.T) {
 		}
 
 		// Close & truncate shadow WAL to simulate a partial frame write.
-		if err := db.Close(); err != nil {
+		if err := db.Close(context.Background()); err != nil {
 			t.Fatal(err)
 		} else if err := os.Truncate(db.ShadowWALPath(pos0.Generation, pos0.Index), fi.Size()-1); err != nil {
 			t.Fatal(err)
@@ -504,7 +503,7 @@ func TestDB_Sync(t *testing.T) {
 		}
 
 		// Close & delete shadow WAL to simulate dir created but not WAL.
-		if err := db.Close(); err != nil {
+		if err := db.Close(context.Background()); err != nil {
 			t.Fatal(err)
 		} else if err := os.Remove(db.ShadowWALPath(pos0.Generation, pos0.Index)); err != nil {
 			t.Fatal(err)
@@ -626,7 +625,7 @@ func MustOpenDBAt(tb testing.TB, path string) *litestream.DB {
 // MustCloseDB closes db and removes its parent directory.
 func MustCloseDB(tb testing.TB, db *litestream.DB) {
 	tb.Helper()
-	if err := db.Close(); err != nil && !strings.Contains(err.Error(), `database is closed`) {
+	if err := db.Close(context.Background()); err != nil && !strings.Contains(err.Error(), `database is closed`) {
 		tb.Fatal(err)
 	} else if err := os.RemoveAll(filepath.Dir(db.Path())); err != nil {
 		tb.Fatal(err)

etc/litestream.yml

@@ -7,4 +7,3 @@
 #    replicas:
 #      - path: /path/to/replica           # File-based replication
 #      - url:  s3://my.bucket.com/db      # S3-based replication
-

etc/s3_mock.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+import sys
+import os
+import time
+from moto.server import ThreadedMotoServer
+import boto3
+import subprocess
+
+cmd = sys.argv[1:]
+if len(cmd) == 0:
+    print(f"usage: {sys.argv[0]} <command> [arguments]", file=sys.stderr)
+    sys.exit(1)
+
+env = os.environ.copy() | {
+    "LITESTREAM_S3_ACCESS_KEY_ID": "lite",
+    "LITESTREAM_S3_SECRET_ACCESS_KEY": "stream",
+    "LITESTREAM_S3_BUCKET": f"test{int(time.time())}",
+    "LITESTREAM_S3_ENDPOINT": "http://127.0.0.1:5000",
+    "LITESTREAM_S3_FORCE_PATH_STYLE": "true",
+}
+
+server = ThreadedMotoServer()
+server.start()
+
+s3 = boto3.client(
+    "s3",
+    aws_access_key_id=env["LITESTREAM_S3_ACCESS_KEY_ID"],
+    aws_secret_access_key=["LITESTREAM_S3_SECRET_ACCESS_KEY"],
+    endpoint_url=env["LITESTREAM_S3_ENDPOINT"]
+).create_bucket(Bucket=env["LITESTREAM_S3_BUCKET"])
+
+proc = subprocess.run(cmd, env=env)
+
+server.stop()
+sys.exit(proc.returncode)

file/replica_client.go

@@ -4,10 +4,8 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"os"
 	"path/filepath"
-	"sort"
 
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/internal"
@@ -112,7 +110,7 @@ func (c *ReplicaClient) Generations(ctx context.Context) ([]string, error) {
 		return nil, fmt.Errorf("cannot determine generations path: %w", err)
 	}
 
-	fis, err := ioutil.ReadDir(root)
+	fis, err := os.ReadDir(root)
 	if os.IsNotExist(err) {
 		return nil, nil
 	} else if err != nil {
@@ -181,8 +179,6 @@ func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (lites
 		})
 	}
 
-	sort.Sort(litestream.SnapshotInfoSlice(infos))
-
 	return litestream.NewSnapshotInfoSliceIterator(infos), nil
 }
 
@@ -298,8 +294,6 @@ func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (lit
 		})
 	}
 
-	sort.Sort(litestream.WALSegmentInfoSlice(infos))
-
 	return litestream.NewWALSegmentInfoSliceIterator(infos), nil
 }
 

go.mod

@@ -1,21 +1,65 @@
 module github.com/benbjohnson/litestream
 
-go 1.16
+go 1.21
 
 require (
-	cloud.google.com/go/storage v1.15.0
+	cloud.google.com/go/storage v1.36.0
-	github.com/Azure/azure-storage-blob-go v0.13.0 // indirect
+	filippo.io/age v1.1.1
-	github.com/Azure/go-autorest/autorest v0.9.0 // indirect
+	github.com/Azure/azure-storage-blob-go v0.15.0
-	github.com/aws/aws-sdk-go v1.27.0
+	github.com/aws/aws-sdk-go v1.49.5
-	github.com/davecgh/go-spew v1.1.1
+	github.com/mattn/go-shellwords v1.0.12
-	github.com/mattn/go-shellwords v1.0.11 // indirect
+	github.com/mattn/go-sqlite3 v1.14.19
-	github.com/mattn/go-sqlite3 v1.14.5
+	github.com/pierrec/lz4/v4 v4.1.19
-	github.com/pierrec/lz4/v4 v4.1.3
+	github.com/pkg/sftp v1.13.6
-	github.com/pkg/sftp v1.13.0 // indirect
+	github.com/prometheus/client_golang v1.17.0
-	github.com/prometheus/client_golang v1.9.0
+	golang.org/x/crypto v0.17.0
-	golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a // indirect
+	golang.org/x/sync v0.5.0
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
+	golang.org/x/sys v0.15.0
-	golang.org/x/sys v0.0.0-20210412220455-f1c623a9e750
+	google.golang.org/api v0.154.0
-	google.golang.org/api v0.45.0
 	gopkg.in/yaml.v2 v2.4.0
 )
+
+require (
+	cloud.google.com/go v0.111.0 // indirect
+	cloud.google.com/go/compute v1.23.3 // indirect
+	cloud.google.com/go/compute/metadata v0.2.3 // indirect
+	cloud.google.com/go/iam v1.1.5 // indirect
+	github.com/Azure/azure-pipeline-go v0.2.3 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
+	github.com/go-logr/logr v1.3.0 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
+	github.com/google/go-cmp v0.6.0 // indirect
+	github.com/google/s2a-go v0.1.7 // indirect
+	github.com/google/uuid v1.5.0 // indirect
+	github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect
+	github.com/googleapis/gax-go/v2 v2.12.0 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
+	github.com/kr/fs v0.1.0 // indirect
+	github.com/mattn/go-ieproxy v0.0.11 // indirect
+	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
+	github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
+	github.com/prometheus/client_model v0.5.0 // indirect
+	github.com/prometheus/common v0.45.0 // indirect
+	github.com/prometheus/procfs v0.12.0 // indirect
+	go.opencensus.io v0.24.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.46.1 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.46.1 // indirect
+	go.opentelemetry.io/otel v1.21.0 // indirect
+	go.opentelemetry.io/otel/metric v1.21.0 // indirect
+	go.opentelemetry.io/otel/trace v1.21.0 // indirect
+	golang.org/x/net v0.19.0 // indirect
+	golang.org/x/oauth2 v0.15.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
+	golang.org/x/time v0.5.0 // indirect
+	golang.org/x/xerrors v0.0.0-20231012003039-104605ab7028 // indirect
+	google.golang.org/appengine v1.6.8 // indirect
+	google.golang.org/genproto v0.0.0-20231212172506-995d672761c0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20231212172506-995d672761c0 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20231212172506-995d672761c0 // indirect
+	google.golang.org/grpc v1.60.1 // indirect
+	google.golang.org/protobuf v1.31.0 // indirect
+)

internal/internal_unix.go

@@ -1,3 +1,4 @@
+//go:build aix || darwin || dragonfly || freebsd || linux || netbsd || openbsd || solaris
 // +build aix darwin dragonfly freebsd linux netbsd openbsd solaris
 
 package internal

internal/internal_windows.go

@@ -1,3 +1,4 @@
+//go:build windows
 // +build windows
 
 package internal

litestream.go

@@ -13,6 +13,8 @@ import (
 	"strconv"
 	"strings"
 	"time"
+
+	"github.com/mattn/go-sqlite3"
 )
 
 // Naming constants.
@@ -42,6 +44,25 @@ var (
 	ErrChecksumMismatch = errors.New("invalid replica, checksum mismatch")
 )
 
+var (
+	// LogWriter is the destination writer for all logging.
+	LogWriter = os.Stdout
+
+	// LogFlags are the flags passed to log.New().
+	LogFlags = 0
+)
+
+func init() {
+	sql.Register("litestream-sqlite3", &sqlite3.SQLiteDriver{
+		ConnectHook: func(conn *sqlite3.SQLiteConn) error {
+			if err := conn.SetFileControlInt("main", sqlite3.SQLITE_FCNTL_PERSIST_WAL, 1); err != nil {
+				return fmt.Errorf("cannot set file control: %w", err)
+			}
+			return nil
+		},
+	})
+}
+
 // SnapshotIterator represents an iterator over a collection of snapshot metadata.
 type SnapshotIterator interface {
 	io.Closer
@@ -207,11 +228,13 @@ func FilterSnapshotsAfter(a []SnapshotInfo, t time.Time) []SnapshotInfo {
 // FindMinSnapshotByGeneration finds the snapshot with the lowest index in a generation.
 func FindMinSnapshotByGeneration(a []SnapshotInfo, generation string) *SnapshotInfo {
 	var min *SnapshotInfo
-	for _, snapshot := range a {
+	for i := range a {
+		snapshot := &a[i]
+
 		if snapshot.Generation != generation {
 			continue
 		} else if min == nil || snapshot.Index < min.Index {
-			min = &snapshot
+			min = snapshot
 		}
 	}
 	return min
@@ -517,9 +540,6 @@ func isHexChar(ch rune) bool {
 	return (ch >= '0' && ch <= '9') || (ch >= 'a' && ch <= 'f')
 }
 
-// Tracef is used for low-level tracing.
-var Tracef = func(format string, a ...interface{}) {}
-
 func assert(condition bool, message string) {
 	if !condition {
 		panic("assertion failed: " + message)

litestream_test.go

@@ -128,6 +128,16 @@ func TestWALSegmentPath(t *testing.T) {
 	})
 }
 
+func TestFindMinSnapshotByGeneration(t *testing.T) {
+	infos := []litestream.SnapshotInfo{
+		{Generation: "29cf4bced74e92ab", Index: 0},
+		{Generation: "5dfeb4aa03232553", Index: 24},
+	}
+	if got, want := litestream.FindMinSnapshotByGeneration(infos, "29cf4bced74e92ab"), &infos[0]; got != want {
+		t.Fatalf("info=%#v, want %#v", got, want)
+	}
+}
+
 func MustDecodeHexString(s string) []byte {
 	b, err := hex.DecodeString(s)
 	if err != nil {

replica.go

@@ -6,8 +6,7 @@ import (
 	"fmt"
 	"hash/crc64"
 	"io"
-	"io/ioutil"
+	"log/slog"
-	"log"
 	"math"
 	"os"
 	"path/filepath"
@@ -15,6 +14,7 @@ import (
 	"sync"
 	"time"
 
+	"filippo.io/age"
 	"github.com/benbjohnson/litestream/internal"
 	"github.com/pierrec/lz4/v4"
 	"github.com/prometheus/client_golang/prometheus"
@@ -67,6 +67,10 @@ type Replica struct {
 	// If true, replica monitors database for changes automatically.
 	// Set to false if replica is being used synchronously (such as in tests).
 	MonitorEnabled bool
+
+	// Encryption identities and recipients
+	AgeIdentities []age.Identity
+	AgeRecipients []age.Recipient
 }
 
 func NewReplica(db *DB, name string) *Replica {
@@ -92,6 +96,15 @@ func (r *Replica) Name() string {
 	return r.name
 }
 
+// Logger returns the DB sub-logger for this replica.
+func (r *Replica) Logger() *slog.Logger {
+	logger := slog.Default()
+	if r.db != nil {
+		logger = r.db.Logger
+	}
+	return logger.With("replica", r.Name())
+}
+
 // DB returns a reference to the database the replica is attached to, if any.
 func (r *Replica) DB() *DB { return r.db }
 
@@ -157,10 +170,13 @@ func (r *Replica) Sync(ctx context.Context) (err error) {
 	}
 	generation := dpos.Generation
 
-	Tracef("%s(%s): replica sync: db.pos=%s", r.db.Path(), r.Name(), dpos)
+	r.Logger().Debug("replica sync", "position", dpos.String())
 
+	// 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)
+		snapshotN, err := r.snapshotN(ctx, generation)
 		if err != nil {
 			return err
 		} else if snapshotN == 0 {
@@ -169,18 +185,14 @@ func (r *Replica) Sync(ctx context.Context) (err error) {
 			} 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.
-	if r.Pos().Generation != generation {
 		pos, err := r.calcPos(ctx, generation)
 		if err != nil {
 			return fmt.Errorf("cannot determine replica position: %s", err)
 		}
 
-		Tracef("%s(%s): replica sync: calc new pos: %s", r.db.Path(), r.Name(), pos)
+		r.Logger().Debug("replica sync: calc new pos", "position", pos.String())
 		r.mu.Lock()
 		r.pos = pos
 		r.mu.Unlock()
@@ -214,16 +226,40 @@ func (r *Replica) syncWAL(ctx context.Context) (err error) {
 	// Obtain initial position from shadow reader.
 	// It may have moved to the next index if previous position was at the end.
 	pos := rd.Pos()
+	initialPos := pos
+	startTime := time.Now()
+	var bytesWritten int
+
+	logger := r.Logger()
+	logger.Info("write wal segment", "position", initialPos.String())
 
 	// Copy through pipe into client from the starting position.
 	var g errgroup.Group
 	g.Go(func() error {
 		_, err := r.Client.WriteWALSegment(ctx, pos, pr)
+
+		// Always close pipe reader to signal writers.
+		if e := pr.CloseWithError(err); err == nil {
+			return e
+		}
+
 		return err
 	})
 
+	var ew io.WriteCloser = pw
+
+	// Add encryption if we have recipients.
+	if len(r.AgeRecipients) > 0 {
+		var err error
+		ew, err = age.Encrypt(pw, r.AgeRecipients...)
+		if err != nil {
+			return err
+		}
+		defer ew.Close()
+	}
+
 	// Wrap writer to LZ4 compress.
-	zw := lz4.NewWriter(pw)
+	zw := lz4.NewWriter(ew)
 
 	// Track total WAL bytes written to replica client.
 	walBytesCounter := replicaWALBytesCounterVec.WithLabelValues(r.db.Path(), r.Name())
@@ -243,6 +279,7 @@ func (r *Replica) syncWAL(ctx context.Context) (err error) {
 			return err
 		}
 		walBytesCounter.Add(float64(n))
+		bytesWritten += n
 	}
 
 	// Copy frames.
@@ -269,11 +306,14 @@ func (r *Replica) syncWAL(ctx context.Context) (err error) {
 			return err
 		}
 		walBytesCounter.Add(float64(n))
+		bytesWritten += n
 	}
 
-	// Flush LZ4 writer and close pipe.
+	// Flush LZ4 writer, encryption writer and close pipe.
 	if err := zw.Close(); err != nil {
 		return err
+	} else if err := ew.Close(); err != nil {
+		return err
 	} else if err := pw.Close(); err != nil {
 		return err
 	}
@@ -292,12 +332,13 @@ func (r *Replica) syncWAL(ctx context.Context) (err error) {
 	replicaWALIndexGaugeVec.WithLabelValues(r.db.Path(), r.Name()).Set(float64(rd.Pos().Index))
 	replicaWALOffsetGaugeVec.WithLabelValues(r.db.Path(), r.Name()).Set(float64(rd.Pos().Offset))
 
+	logger.Info("wal segment written", "position", initialPos.String(), "elapsed", time.Since(startTime).String(), "sz", bytesWritten)
 	return nil
 }
 
 // snapshotN returns the number of snapshots for a generation.
-func (r *Replica) snapshotN(generation string) (int, error) {
+func (r *Replica) snapshotN(ctx context.Context, generation string) (int, error) {
-	itr, err := r.Client.Snapshots(context.Background(), generation)
+	itr, err := r.Client.Snapshots(ctx, generation)
 	if err != nil {
 		return 0, err
 	}
@@ -335,7 +376,16 @@ func (r *Replica) calcPos(ctx context.Context, generation string) (pos Pos, err
 	}
 	defer rd.Close()
 
-	n, err := io.Copy(ioutil.Discard, lz4.NewReader(rd))
+	if len(r.AgeIdentities) > 0 {
+		drd, err := age.Decrypt(rd, r.AgeIdentities...)
+		if err != nil {
+			return pos, err
+		}
+
+		rd = io.NopCloser(drd)
+	}
+
+	n, err := io.Copy(io.Discard, lz4.NewReader(rd))
 	if err != nil {
 		return pos, err
 	}
@@ -433,11 +483,15 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
 	defer r.muf.Unlock()
 
 	// Issue a passive checkpoint to flush any pages to disk before snapshotting.
-	if _, err := r.db.db.ExecContext(ctx, `PRAGMA wal_checkpoint(PASSIVE);`); err != nil {
+	if err := r.db.Checkpoint(ctx, CheckpointModePassive); err != nil {
 		return info, fmt.Errorf("pre-snapshot checkpoint: %w", err)
 	}
 
-	// Acquire a read lock on the database during snapshot to prevent checkpoints.
+	// Prevent internal checkpoints during snapshot.
+	r.db.BeginSnapshot()
+	defer r.db.EndSnapshot()
+
+	// Acquire a read lock on the database during snapshot to prevent external checkpoints.
 	tx, err := r.db.db.Begin()
 	if err != nil {
 		return info, err
@@ -471,7 +525,23 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
 	// Copy the database file to the LZ4 writer in a separate goroutine.
 	var g errgroup.Group
 	g.Go(func() error {
-		zr := lz4.NewWriter(pw)
+		// We need to ensure the pipe is closed.
+		defer pw.Close()
+
+		var wc io.WriteCloser = pw
+
+		// Add encryption if we have recipients.
+		if len(r.AgeRecipients) > 0 {
+			var err error
+			wc, err = age.Encrypt(pw, r.AgeRecipients...)
+			if err != nil {
+				pw.CloseWithError(err)
+				return err
+			}
+			defer wc.Close()
+		}
+
+		zr := lz4.NewWriter(wc)
 		defer zr.Close()
 
 		if _, err := io.Copy(zr, r.f); err != nil {
@@ -481,9 +551,13 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
 			pw.CloseWithError(err)
 			return err
 		}
-		return pw.Close()
+		return wc.Close()
 	})
 
+	logger := r.Logger()
+	logger.Info("write snapshot", "position", pos.String())
+
+	startTime := time.Now()
 	// Delegate write to client & wait for writer goroutine to finish.
 	if info, err = r.Client.WriteSnapshot(ctx, pos.Generation, pos.Index, pr); err != nil {
 		return info, err
@@ -491,8 +565,7 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
 		return info, err
 	}
 
-	log.Printf("%s(%s): snapshot written %s/%08x", r.db.Path(), r.Name(), pos.Generation, pos.Index)
+	logger.Info("snapshot written", "position", pos.String(), "elapsed", time.Since(startTime).String(), "sz", info.Size)
-
 	return info, nil
 }
 
@@ -559,7 +632,7 @@ func (r *Replica) deleteSnapshotsBeforeIndex(ctx context.Context, generation str
 		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)
 		}
-		log.Printf("%s(%s): snapshot deleted %s/%08x", r.db.Path(), r.Name(), generation, index)
+		r.Logger().Info("snapshot deleted", "generation", generation, "index", index)
 	}
 
 	return itr.Close()
@@ -591,8 +664,8 @@ func (r *Replica) deleteWALSegmentsBeforeIndex(ctx context.Context, generation s
 	if err := r.Client.DeleteWALSegments(ctx, a); err != nil {
 		return fmt.Errorf("delete wal segments: %w", err)
 	}
-	log.Printf("%s(%s): wal segmented deleted before %s/%08x: n=%d", r.db.Path(), r.Name(), generation, index, len(a))
 
+	r.Logger().Info("wal segmented deleted before", "generation", generation, "index", index, "n", len(a))
 	return nil
 }
 
@@ -628,7 +701,7 @@ func (r *Replica) monitor(ctx context.Context) {
 
 		// Synchronize the shadow wal into the replication directory.
 		if err := r.Sync(ctx); err != nil {
-			log.Printf("%s(%s): monitor error: %s", r.db.Path(), r.Name(), err)
+			r.Logger().Error("monitor error", "error", err)
 			continue
 		}
 	}
@@ -656,7 +729,7 @@ func (r *Replica) retainer(ctx context.Context) {
 			return
 		case <-ticker.C:
 			if err := r.EnforceRetention(ctx); err != nil {
-				log.Printf("%s(%s): retainer error: %s", r.db.Path(), r.Name(), err)
+				r.Logger().Error("retainer error", "error", err)
 				continue
 			}
 		}
@@ -669,6 +742,31 @@ func (r *Replica) snapshotter(ctx context.Context) {
 		return
 	}
 
+	logger := r.Logger()
+	if pos, err := r.db.Pos(); err != nil {
+		logger.Error("snapshotter cannot determine generation", "error", err)
+	} else if !pos.IsZero() {
+		if snapshot, err := r.maxSnapshot(ctx, pos.Generation); err != nil {
+			logger.Error("snapshotter cannot determine latest snapshot", "error", err)
+		} else if snapshot != nil {
+			nextSnapshot := r.SnapshotInterval - time.Since(snapshot.CreatedAt)
+			if nextSnapshot < 0 {
+				nextSnapshot = 0
+			}
+
+			logger.Info("snapshot interval adjusted", "previous", snapshot.CreatedAt.Format(time.RFC3339), "next", nextSnapshot.String())
+
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(nextSnapshot):
+				if _, err := r.Snapshot(ctx); err != nil && err != ErrNoGeneration {
+					logger.Error("snapshotter error", "error", err)
+				}
+			}
+		}
+	}
+
 	ticker := time.NewTicker(r.SnapshotInterval)
 	defer ticker.Stop()
 
@@ -678,7 +776,7 @@ func (r *Replica) snapshotter(ctx context.Context) {
 			return
 		case <-ticker.C:
 			if _, err := r.Snapshot(ctx); err != nil && err != ErrNoGeneration {
-				log.Printf("%s(%s): snapshotter error: %s", r.db.Path(), r.Name(), err)
+				r.Logger().Error("snapshotter error", "error", err)
 				continue
 			}
 		}
@@ -706,7 +804,7 @@ func (r *Replica) validator(ctx context.Context) {
 			return
 		case <-ticker.C:
 			if err := r.Validate(ctx); err != nil {
-				log.Printf("%s(%s): validation error: %s", r.db.Path(), r.Name(), err)
+				r.Logger().Error("validation error", "error", err)
 				continue
 			}
 		}
@@ -719,7 +817,7 @@ func (r *Replica) Validate(ctx context.Context) error {
 	db := r.DB()
 
 	// Restore replica to a temporary directory.
-	tmpdir, err := ioutil.TempDir("", "*-litestream")
+	tmpdir, err := os.MkdirTemp("", "*-litestream")
 	if err != nil {
 		return err
 	}
@@ -743,7 +841,6 @@ func (r *Replica) Validate(ctx context.Context) error {
 		ReplicaName: r.Name(),
 		Generation:  pos.Generation,
 		Index:       pos.Index - 1,
-		Logger:      log.New(os.Stderr, "", 0),
 	}); err != nil {
 		return fmt.Errorf("cannot restore: %w", err)
 	}
@@ -768,7 +865,7 @@ func (r *Replica) Validate(ctx context.Context) error {
 	if mismatch {
 		status = "mismatch"
 	}
-	log.Printf("%s(%s): validator: status=%s db=%016x replica=%016x pos=%s", db.Path(), r.Name(), status, chksum0, chksum1, pos)
+	r.Logger().Info("validator", "status", status, "db", fmt.Sprintf("%016x", chksum0), "replica", fmt.Sprintf("%016x", chksum1), "position", pos.String())
 
 	// Validate checksums match.
 	if mismatch {
@@ -786,8 +883,6 @@ func (r *Replica) Validate(ctx context.Context) error {
 
 // waitForReplica blocks until replica reaches at least the given position.
 func (r *Replica) waitForReplica(ctx context.Context, pos Pos) error {
-	db := r.DB()
-
 	ticker := time.NewTicker(500 * time.Millisecond)
 	defer ticker.Stop()
 
@@ -810,7 +905,7 @@ func (r *Replica) waitForReplica(ctx context.Context, pos Pos) error {
 		// Obtain current position of replica, check if past target position.
 		curr := r.Pos()
 		if curr.IsZero() {
-			log.Printf("%s(%s): validator: no replica position available", db.Path(), r.Name())
+			r.Logger().Info("validator: no replica position available")
 			continue
 		}
 
@@ -866,6 +961,7 @@ func (r *Replica) GenerationTimeBounds(ctx context.Context, generation string) (
 	}
 	defer sitr.Close()
 
+	minIndex, maxIndex := -1, -1
 	for sitr.Next() {
 		info := sitr.Snapshot()
 		if createdAt.IsZero() || info.CreatedAt.Before(createdAt) {
@@ -874,6 +970,12 @@ func (r *Replica) GenerationTimeBounds(ctx context.Context, generation string) (
 		if updatedAt.IsZero() || info.CreatedAt.After(updatedAt) {
 			updatedAt = info.CreatedAt
 		}
+		if minIndex == -1 || info.Index < minIndex {
+			minIndex = info.Index
+		}
+		if info.Index > maxIndex {
+			maxIndex = info.Index
+		}
 	}
 	if err := sitr.Close(); err != nil {
 		return createdAt, updatedAt, err
@@ -888,6 +990,9 @@ func (r *Replica) GenerationTimeBounds(ctx context.Context, generation string) (
 
 	for witr.Next() {
 		info := witr.WALSegment()
+		if info.Index < minIndex || info.Index > maxIndex {
+			continue
+		}
 		if createdAt.IsZero() || info.CreatedAt.Before(createdAt) {
 			createdAt = info.CreatedAt
 		}
@@ -962,17 +1067,6 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 		return fmt.Errorf("cannot specify index & timestamp to restore")
 	}
 
-	// Ensure logger exists.
-	logger := opt.Logger
-	if logger == nil {
-		logger = log.New(ioutil.Discard, "", 0)
-	}
-
-	logPrefix := r.Name()
-	if db := r.DB(); db != nil {
-		logPrefix = fmt.Sprintf("%s(%s)", db.Path(), r.Name())
-	}
-
 	// Ensure output path does not already exist.
 	if _, err := os.Stat(opt.OutputPath); err == nil {
 		return fmt.Errorf("cannot restore, output path already exists: %s", opt.OutputPath)
@@ -993,7 +1087,7 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 	}
 
 	// Compute list of offsets for each WAL index.
-	walSegmentMap, err := r.walSegmentMap(ctx, opt.Generation, opt.Index, opt.Timestamp)
+	walSegmentMap, err := r.walSegmentMap(ctx, opt.Generation, minWALIndex, opt.Index, opt.Timestamp)
 	if err != nil {
 		return fmt.Errorf("cannot find max wal index for restore: %w", err)
 	}
@@ -1007,7 +1101,7 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 	}
 
 	// Ensure that we found the specific index, if one was specified.
-	if opt.Index != math.MaxInt32 && opt.Index != opt.Index {
+	if opt.Index != math.MaxInt32 && opt.Index != maxWALIndex {
 		return fmt.Errorf("unable to locate index %d in generation %q, highest index was %d", opt.Index, opt.Generation, maxWALIndex)
 	}
 
@@ -1019,19 +1113,19 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 	tmpPath := opt.OutputPath + ".tmp"
 
 	// Copy snapshot to output path.
-	logger.Printf("%s: restoring snapshot %s/%08x to %s", logPrefix, opt.Generation, minWALIndex, tmpPath)
+	r.Logger().Info("restoring snapshot", "generation", opt.Generation, "index", minWALIndex, "path", tmpPath)
 	if err := r.restoreSnapshot(ctx, pos.Generation, pos.Index, tmpPath); err != nil {
 		return fmt.Errorf("cannot restore snapshot: %w", err)
 	}
 
 	// If no WAL files available, move snapshot to final path & exit early.
 	if snapshotOnly {
-		logger.Printf("%s: snapshot only, finalizing database", logPrefix)
+		r.Logger().Info("snapshot only, finalizing database")
 		return os.Rename(tmpPath, opt.OutputPath)
 	}
 
 	// Begin processing WAL files.
-	logger.Printf("%s: restoring wal files: generation=%s index=[%08x,%08x]", logPrefix, opt.Generation, minWALIndex, maxWALIndex)
+	r.Logger().Info("restoring wal files", "generation", opt.Generation, "index_min", minWALIndex, "index_max", maxWALIndex)
 
 	// Fill input channel with all WAL indexes to be loaded in order.
 	// Verify every index has at least one offset.
@@ -1087,9 +1181,9 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 						return err
 					}
 
-					logger.Printf("%s: downloaded wal %s/%08x elapsed=%s",
+					r.Logger().Info("downloaded wal",
-						logPrefix, opt.Generation, index,
+						"generation", opt.Generation, "index", index,
-						time.Since(startTime).String(),
+						"elapsed", time.Since(startTime).String(),
 					)
 				}
 			}
@@ -1116,10 +1210,7 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 		if err = applyWAL(ctx, index, tmpPath); err != nil {
 			return fmt.Errorf("cannot apply wal: %w", err)
 		}
-		logger.Printf("%s: applied wal %s/%08x elapsed=%s",
+		r.Logger().Info("applied wal", "generation", opt.Generation, "index", index, "elapsed", time.Since(startTime).String())
-			logPrefix, opt.Generation, index,
-			time.Since(startTime).String(),
-		)
 	}
 
 	// Ensure all goroutines finish. All errors should have been handled during
@@ -1129,7 +1220,7 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 	}
 
 	// Copy file to final location.
-	logger.Printf("%s: renaming database from temporary location", logPrefix)
+	r.Logger().Info("renaming database from temporary location")
 	if err := os.Rename(tmpPath, opt.OutputPath); err != nil {
 		return err
 	}
@@ -1190,7 +1281,7 @@ func (r *Replica) SnapshotIndexByIndex(ctx context.Context, generation string, i
 		}
 
 		// Use snapshot if it newer.
-		if snapshotIndex == -1 || snapshotIndex >= snapshotIndex {
+		if snapshotIndex == -1 || snapshot.Index >= snapshotIndex {
 			snapshotIndex = snapshot.Index
 		}
 	}
@@ -1204,22 +1295,29 @@ func (r *Replica) SnapshotIndexByIndex(ctx context.Context, generation string, i
 
 // walSegmentMap returns a map of WAL indices to their segments.
 // Filters by a max timestamp or a max index.
-func (r *Replica) walSegmentMap(ctx context.Context, generation string, maxIndex int, maxTimestamp time.Time) (map[int][]int64, error) {
+func (r *Replica) walSegmentMap(ctx context.Context, generation string, minIndex, maxIndex int, maxTimestamp time.Time) (map[int][]int64, error) {
 	itr, err := r.Client.WALSegments(ctx, generation)
 	if err != nil {
 		return nil, err
 	}
 	defer itr.Close()
 
-	m := make(map[int][]int64)
+	a := []WALSegmentInfo{}
 	for itr.Next() {
-		info := itr.WALSegment()
+		a = append(a, itr.WALSegment())
+	}
 
+	sort.Sort(WALSegmentInfoSlice(a))
+
+	m := make(map[int][]int64)
+	for _, info := range a {
 		// Exit if we go past the max timestamp or index.
 		if !maxTimestamp.IsZero() && info.CreatedAt.After(maxTimestamp) {
 			break // after max timestamp, skip
 		} else if info.Index > maxIndex {
 			break // after max index, skip
+		} else if info.Index < minIndex {
+			continue // before min index, continue
 		}
 
 		// Verify offsets are added in order.
@@ -1260,6 +1358,15 @@ func (r *Replica) restoreSnapshot(ctx context.Context, generation string, index
 	}
 	defer rd.Close()
 
+	if len(r.AgeIdentities) > 0 {
+		drd, err := age.Decrypt(rd, r.AgeIdentities...)
+		if err != nil {
+			return err
+		}
+
+		rd = io.NopCloser(drd)
+	}
+
 	if _, err := io.Copy(f, lz4.NewReader(rd)); err != nil {
 		return err
 	} else if err := f.Sync(); err != nil {
@@ -1286,6 +1393,16 @@ func (r *Replica) downloadWAL(ctx context.Context, generation string, index int,
 			return err
 		}
 		defer rd.Close()
+
+		if len(r.AgeIdentities) > 0 {
+			drd, err := age.Decrypt(rd, r.AgeIdentities...)
+			if err != nil {
+				return err
+			}
+
+			rd = io.NopCloser(drd)
+		}
+
 		readers = append(readers, lz4.NewReader(rd))
 	}
 
@@ -1307,13 +1424,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.go

@@ -10,13 +10,13 @@ type ReplicaClient interface {
 	// Returns the type of client.
 	Type() string
 
-	// Returns a list of available generations.
+	// Returns a list of available generations. Order is undefined.
 	Generations(ctx context.Context) ([]string, error)
 
 	// Deletes all snapshots & WAL segments within a generation.
 	DeleteGeneration(ctx context.Context, generation string) error
 
-	// Returns an iterator of all snapshots within a generation on the replica.
+	// Returns an iterator of all snapshots within a generation on the replica. Order is undefined.
 	Snapshots(ctx context.Context, generation string) (SnapshotIterator, error)
 
 	// Writes LZ4 compressed snapshot data to the replica at a given index
@@ -31,7 +31,7 @@ type ReplicaClient interface {
 	// the snapshot does not exist.
 	SnapshotReader(ctx context.Context, generation string, index int) (io.ReadCloser, error)
 
-	// Returns an iterator of all WAL segments within a generation on the replica.
+	// Returns an iterator of all WAL segments within a generation on the replica. Order is undefined.
 	WALSegments(ctx context.Context, generation string) (WALSegmentIterator, error)
 
 	// Writes an LZ4 compressed WAL segment at a given position.

replica_client_test.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"io/ioutil"
+	"io"
 	"math/rand"
 	"os"
 	"path"
@@ -12,7 +12,6 @@ import (
 	"sort"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/abs"
@@ -22,10 +21,6 @@ import (
 	"github.com/benbjohnson/litestream/sftp"
 )
 
-func init() {
-	rand.Seed(time.Now().UnixNano())
-}
-
 var (
 	// Enables integration tests.
 	integration = flag.String("integration", "file", "")
@@ -80,10 +75,14 @@ func TestReplicaClient_Generations(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		// Verify returned generations.
+		// Fetch and sort generations.
-		if got, err := c.Generations(context.Background()); err != nil {
+		got, err := c.Generations(context.Background())
+		if err != nil {
 			t.Fatal(err)
-		} else if want := []string{"155fe292f8333c72", "5efbd8d042012dca", "b16ddcf5c697540f"}; !reflect.DeepEqual(got, want) {
+		}
+		sort.Strings(got)
+
+		if want := []string{"155fe292f8333c72", "5efbd8d042012dca", "b16ddcf5c697540f"}; !reflect.DeepEqual(got, want) {
 			t.Fatalf("Generations()=%v, want %v", got, want)
 		}
 	})
@@ -193,7 +192,7 @@ func TestReplicaClient_WriteSnapshot(t *testing.T) {
 
 		if r, err := c.SnapshotReader(context.Background(), "b16ddcf5c697540f", 1000); err != nil {
 			t.Fatal(err)
-		} else if buf, err := ioutil.ReadAll(r); err != nil {
+		} else if buf, err := io.ReadAll(r); err != nil {
 			t.Fatal(err)
 		} else if err := r.Close(); err != nil {
 			t.Fatal(err)
@@ -224,7 +223,7 @@ func TestReplicaClient_SnapshotReader(t *testing.T) {
 		}
 		defer r.Close()
 
-		if buf, err := ioutil.ReadAll(r); err != nil {
+		if buf, err := io.ReadAll(r); err != nil {
 			t.Fatal(err)
 		} else if got, want := string(buf), "foo"; got != want {
 			t.Fatalf("ReadAll=%v, want %v", got, want)
@@ -378,7 +377,7 @@ func TestReplicaClient_WriteWALSegment(t *testing.T) {
 
 		if r, err := c.WALSegmentReader(context.Background(), litestream.Pos{Generation: "b16ddcf5c697540f", Index: 1000, Offset: 2000}); err != nil {
 			t.Fatal(err)
-		} else if buf, err := ioutil.ReadAll(r); err != nil {
+		} else if buf, err := io.ReadAll(r); err != nil {
 			t.Fatal(err)
 		} else if err := r.Close(); err != nil {
 			t.Fatal(err)
@@ -409,7 +408,7 @@ func TestReplicaClient_WALReader(t *testing.T) {
 		}
 		defer r.Close()
 
-		if buf, err := ioutil.ReadAll(r); err != nil {
+		if buf, err := io.ReadAll(r); err != nil {
 			t.Fatal(err)
 		} else if got, want := string(buf), "foobar"; got != want {
 			t.Fatalf("ReadAll=%v, want %v", got, want)

replica_test.go

@@ -13,6 +13,13 @@ import (
 	"github.com/pierrec/lz4/v4"
 )
 
+func nextIndex(pos litestream.Pos) litestream.Pos {
+	return litestream.Pos{
+		Generation: pos.Generation,
+		Index:      pos.Index + 1,
+	}
+}
+
 func TestReplica_Name(t *testing.T) {
 	t.Run("WithName", func(t *testing.T) {
 		if got, want := litestream.NewReplica(nil, "NAME").Name(), "NAME"; got != want {
@@ -32,11 +39,6 @@ func TestReplica_Sync(t *testing.T) {
 	db, sqldb := MustOpenDBs(t)
 	defer MustCloseDBs(t, db, sqldb)
 
-	// Execute a query to force a write to the WAL.
-	if _, err := sqldb.Exec(`CREATE TABLE foo (bar TEXT);`); err != nil {
-		t.Fatal(err)
-	}
-
 	// Issue initial database sync to setup generation.
 	if err := db.Sync(context.Background()); err != nil {
 		t.Fatal(err)
@@ -66,10 +68,47 @@ func TestReplica_Sync(t *testing.T) {
 		t.Fatalf("generations[0]=%v, want %v", got, want)
 	}
 
+	// Verify we synced checkpoint page to WAL.
+	if r, err := c.WALSegmentReader(context.Background(), nextIndex(dpos)); err != nil {
+		t.Fatal(err)
+	} else if b, err := io.ReadAll(lz4.NewReader(r)); err != nil {
+		t.Fatal(err)
+	} else if err := r.Close(); err != nil {
+		t.Fatal(err)
+	} else if len(b) == db.PageSize() {
+		t.Fatalf("wal mismatch: len(%d), len(%d)", len(b), db.PageSize())
+	}
+
+	// Reset WAL so the next write will only write out the segment we are checking.
+	if err := db.Checkpoint(context.Background(), litestream.CheckpointModeTruncate); err != nil {
+		t.Fatal(err)
+	}
+
+	// Execute a query to write something into the truncated WAL.
+	if _, err := sqldb.Exec(`CREATE TABLE foo (bar TEXT);`); err != nil {
+		t.Fatal(err)
+	}
+
+	// Sync database to catch up the shadow WAL.
+	if err := db.Sync(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+
+	// Save position after sync, it should be after our write.
+	dpos, err = db.Pos()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Sync WAL segment out to replica.
+	if err := r.Sync(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+
 	// Verify WAL matches replica WAL.
 	if b0, err := os.ReadFile(db.Path() + "-wal"); err != nil {
 		t.Fatal(err)
-	} else if r, err := c.WALSegmentReader(context.Background(), litestream.Pos{Generation: generations[0], Index: 0, Offset: 0}); err != nil {
+	} else if r, err := c.WALSegmentReader(context.Background(), dpos.Truncate()); err != nil {
 		t.Fatal(err)
 	} else if b1, err := io.ReadAll(lz4.NewReader(r)); err != nil {
 		t.Fatal(err)
@@ -103,7 +142,7 @@ func TestReplica_Snapshot(t *testing.T) {
 		t.Fatal(err)
 	} else if info, err := r.Snapshot(context.Background()); err != nil {
 		t.Fatal(err)
-	} else if got, want := info.Pos(), pos0.Truncate(); got != want {
+	} else if got, want := info.Pos(), nextIndex(pos0); got != want {
 		t.Fatalf("pos=%s, want %s", got, want)
 	}
 
@@ -127,18 +166,20 @@ func TestReplica_Snapshot(t *testing.T) {
 		t.Fatal(err)
 	} else if info, err := r.Snapshot(context.Background()); err != nil {
 		t.Fatal(err)
-	} else if got, want := info.Pos(), pos1.Truncate(); got != want {
+	} else if got, want := info.Pos(), nextIndex(pos1); got != want {
 		t.Fatalf("pos=%v, want %v", got, want)
 	}
 
-	// Verify two snapshots exist.
+	// Verify three snapshots exist.
 	if infos, err := r.Snapshots(context.Background()); err != nil {
 		t.Fatal(err)
-	} else if got, want := len(infos), 2; got != want {
+	} else if got, want := len(infos), 3; got != want {
 		t.Fatalf("len=%v, want %v", got, want)
 	} else if got, want := infos[0].Pos(), pos0.Truncate(); got != want {
 		t.Fatalf("info[0]=%s, want %s", got, want)
-	} else if got, want := infos[1].Pos(), pos1.Truncate(); got != want {
+	} else if got, want := infos[1].Pos(), nextIndex(pos0); got != want {
 		t.Fatalf("info[1]=%s, want %s", got, want)
+	} else if got, want := infos[2].Pos(), nextIndex(pos1); got != want {
+		t.Fatalf("info[2]=%s, want %s", got, want)
 	}
 }

s3/replica_client.go

@@ -16,7 +16,6 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/credentials"
-	"github.com/aws/aws-sdk-go/aws/defaults"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/s3/s3manager"
@@ -93,6 +92,7 @@ func (c *ReplicaClient) Init(ctx context.Context) (err error) {
 	if region != "" {
 		config.Region = aws.String(region)
 	}
+
 	sess, err := session.NewSession(config)
 	if err != nil {
 		return fmt.Errorf("cannot create aws session: %w", err)
@@ -105,7 +105,8 @@ func (c *ReplicaClient) Init(ctx context.Context) (err error) {
 // config returns the AWS configuration. Uses the default credential chain
 // unless a key/secret are explicitly set.
 func (c *ReplicaClient) config() *aws.Config {
-	config := defaults.Get().Config
+	config := &aws.Config{}
+
 	if c.AccessKeyID != "" || c.SecretAccessKey != "" {
 		config.Credentials = credentials.NewStaticCredentials(c.AccessKeyID, c.SecretAccessKey, "")
 	}
@@ -135,12 +136,12 @@ func (c *ReplicaClient) findBucketRegion(ctx context.Context, bucket string) (st
 
 	// Fetch bucket location, if possible. Must be bucket owner.
 	// This call can return a nil location which means it's in us-east-1.
-	if out, err := s3.New(sess).GetBucketLocation(&s3.GetBucketLocationInput{
+	if out, err := s3.New(sess).HeadBucketWithContext(ctx, &s3.HeadBucketInput{
 		Bucket: aws.String(bucket),
 	}); err != nil {
 		return "", err
-	} else if out.LocationConstraint != nil {
+	} else if out.BucketRegion != nil {
-		return *out.LocationConstraint, nil
+		return *out.BucketRegion, nil
 	}
 	return DefaultRegion, nil
 }
@@ -160,7 +161,7 @@ func (c *ReplicaClient) Generations(ctx context.Context) ([]string, error) {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
 		for _, prefix := range page.CommonPrefixes {
-			name := path.Base(*prefix.Prefix)
+			name := path.Base(aws.StringValue(prefix.Prefix))
 			if !litestream.IsGenerationName(name) {
 				continue
 			}
@@ -208,10 +209,14 @@ func (c *ReplicaClient) DeleteGeneration(ctx context.Context, generation string)
 			n = len(objIDs)
 		}
 
-		if _, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
+		out, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
 			Bucket: aws.String(c.Bucket),
 			Delete: &s3.Delete{Objects: objIDs[:n], Quiet: aws.Bool(true)},
-		}); err != nil {
+		})
+		if err != nil {
+			return err
+		}
+		if err := deleteOutputError(out); err != nil {
 			return err
 		}
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "DELETE").Inc()
@@ -287,7 +292,7 @@ func (c *ReplicaClient) SnapshotReader(ctx context.Context, generation string, i
 		return nil, err
 	}
 	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "GET").Inc()
-	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "GET").Add(float64(*out.ContentLength))
+	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "GET").Add(float64(aws.Int64Value(out.ContentLength)))
 
 	return out.Body, nil
 }
@@ -303,10 +308,14 @@ func (c *ReplicaClient) DeleteSnapshot(ctx context.Context, generation string, i
 		return fmt.Errorf("cannot determine snapshot path: %w", err)
 	}
 
-	if _, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
+	out, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
 		Bucket: aws.String(c.Bucket),
 		Delete: &s3.Delete{Objects: []*s3.ObjectIdentifier{{Key: &key}}, Quiet: aws.Bool(true)},
-	}); err != nil {
+	})
+	if err != nil {
+		return err
+	}
+	if err := deleteOutputError(out); err != nil {
 		return err
 	}
 
@@ -377,7 +386,7 @@ func (c *ReplicaClient) WALSegmentReader(ctx context.Context, pos litestream.Pos
 		return nil, err
 	}
 	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "GET").Inc()
-	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "GET").Add(float64(*out.ContentLength))
+	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "GET").Add(float64(aws.Int64Value(out.ContentLength)))
 
 	return out.Body, nil
 }
@@ -405,13 +414,16 @@ func (c *ReplicaClient) DeleteWALSegments(ctx context.Context, a []litestream.Po
 		}
 
 		// Delete S3 objects in bulk.
-		if _, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
+		out, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
 			Bucket: aws.String(c.Bucket),
 			Delete: &s3.Delete{Objects: objIDs[:n], Quiet: aws.Bool(true)},
-		}); err != nil {
+		})
+		if err != nil {
+			return err
+		}
+		if err := deleteOutputError(out); err != nil {
 			return err
 		}
-
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "DELETE").Inc()
 
 		a = a[n:]
@@ -454,10 +466,14 @@ func (c *ReplicaClient) DeleteAll(ctx context.Context) error {
 			n = len(objIDs)
 		}
 
-		if _, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
+		out, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
 			Bucket: aws.String(c.Bucket),
 			Delete: &s3.Delete{Objects: objIDs[:n], Quiet: aws.Bool(true)},
-		}); err != nil {
+		})
+		if err != nil {
+			return err
+		}
+		if err := deleteOutputError(out); err != nil {
 			return err
 		}
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "DELETE").Inc()
@@ -511,7 +527,7 @@ func (itr *snapshotIterator) fetch() error {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
 		for _, obj := range page.Contents {
-			key := path.Base(*obj.Key)
+			key := path.Base(aws.StringValue(obj.Key))
 			index, err := litestream.ParseSnapshotPath(key)
 			if err != nil {
 				continue
@@ -520,7 +536,7 @@ func (itr *snapshotIterator) fetch() error {
 			info := litestream.SnapshotInfo{
 				Generation: itr.generation,
 				Index:      index,
-				Size:       *obj.Size,
+				Size:       aws.Int64Value(obj.Size),
 				CreatedAt:  obj.LastModified.UTC(),
 			}
 
@@ -614,7 +630,7 @@ func (itr *walSegmentIterator) fetch() error {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
 		for _, obj := range page.Contents {
-			key := path.Base(*obj.Key)
+			key := path.Base(aws.StringValue(obj.Key))
 			index, offset, err := litestream.ParseWALSegmentPath(key)
 			if err != nil {
 				continue
@@ -624,7 +640,7 @@ func (itr *walSegmentIterator) fetch() error {
 				Generation: itr.generation,
 				Index:      index,
 				Offset:     offset,
-				Size:       *obj.Size,
+				Size:       aws.Int64Value(obj.Size),
 				CreatedAt:  obj.LastModified.UTC(),
 			}
 
@@ -700,6 +716,9 @@ func ParseHost(s string) (bucket, region, endpoint string, forcePathStyle bool)
 	} else if a := digitalOceanRegex.FindStringSubmatch(host); a != nil {
 		bucket, region = a[1], a[2]
 		endpoint = fmt.Sprintf("%s.digitaloceanspaces.com", region)
+	} else if a := scalewayRegex.FindStringSubmatch(host); a != nil {
+		bucket, region = a[1], a[2]
+		endpoint = fmt.Sprintf("s3.%s.scw.cloud", region)
 	} else if a := linodeRegex.FindStringSubmatch(host); a != nil {
 		bucket, region = a[1], a[2]
 		endpoint = fmt.Sprintf("%s.linodeobjects.com", region)
@@ -726,6 +745,7 @@ var (
 	backblazeRegex    = regexp.MustCompile(`^(?:(.+)\.)?s3.([^.]+)\.backblazeb2.com$`)
 	filebaseRegex     = regexp.MustCompile(`^(?:(.+)\.)?s3.filebase.com$`)
 	digitalOceanRegex = regexp.MustCompile(`^(?:(.+)\.)?([^.]+)\.digitaloceanspaces.com$`)
+	scalewayRegex     = regexp.MustCompile(`^(?:(.+)\.)?s3.([^.]+)\.scw\.cloud$`)
 	linodeRegex       = regexp.MustCompile(`^(?:(.+)\.)?([^.]+)\.linodeobjects.com$`)
 )
 
@@ -737,3 +757,15 @@ func isNotExists(err error) bool {
 		return false
 	}
 }
+
+func deleteOutputError(out *s3.DeleteObjectsOutput) error {
+	switch len(out.Errors) {
+	case 0:
+		return nil
+	case 1:
+		return fmt.Errorf("deleting object %s: %s - %s", aws.StringValue(out.Errors[0].Key), aws.StringValue(out.Errors[0].Code), aws.StringValue(out.Errors[0].Message))
+	default:
+		return fmt.Errorf("%d errors occurred deleting objects, %s: %s - (%s (and %d others)",
+			len(out.Errors), aws.StringValue(out.Errors[0].Key), aws.StringValue(out.Errors[0].Code), aws.StringValue(out.Errors[0].Message), len(out.Errors)-1)
+	}
+}

sftp/replica_client.go

@@ -8,7 +8,6 @@ import (
 	"net"
 	"os"
 	"path"
-	"sort"
 	"sync"
 	"time"
 
@@ -141,8 +140,6 @@ func (c *ReplicaClient) Generations(ctx context.Context) (_ []string, err error)
 		generations = append(generations, name)
 	}
 
-	sort.Strings(generations)
-
 	return generations, nil
 }
 
@@ -229,8 +226,6 @@ func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (_ lit
 		})
 	}
 
-	sort.Sort(litestream.SnapshotInfoSlice(infos))
-
 	return litestream.NewSnapshotInfoSliceIterator(infos), nil
 }
 
@@ -363,8 +358,6 @@ func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (_ l
 		})
 	}
 
-	sort.Sort(litestream.WALSegmentInfoSlice(infos))
-
 	return litestream.NewWALSegmentInfoSliceIterator(infos), nil
 }