Reduce snapshot check frequency

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

.github/CONTRIBUTING.md

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

.github/pull_request_template.md

@@ -0,0 +1,7 @@
+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

@@ -1,229 +0,0 @@
-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

@@ -1,51 +0,0 @@
-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-latest
+    runs-on: ubuntu-18.04
     strategy:
       matrix:
         include:
@@ -31,7 +31,7 @@ jobs:
       - uses: actions/checkout@v2
       - uses: actions/setup-go@v2
         with:
-          go-version: '1.21'
+          go-version: '1.16'
 
       - id: release
         uses: bruceadams/get-release@v1.2.2
@@ -40,7 +40,7 @@ jobs:
 
       - name: Install cross-compilers
         run: |
-          sudo apt-get update
+          sudo apt-get update 
           sudo apt-get install -y gcc-aarch64-linux-gnu gcc-arm-linux-gnueabihf gcc-arm-linux-gnueabi
 
       - name: Install nfpm
@@ -54,11 +54,11 @@ 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 -extldflags "-static" -X 'main.Version=${{ steps.release.outputs.tag_name }}'" -tags osusergo,netgo,sqlite_omit_load_extension -o dist/litestream ./cmd/litestream
+          CGO_ENABLED=1 go build -ldflags "-s -w -X 'main.Version=${{ steps.release.outputs.tag_name }}'" -o dist/litestream ./cmd/litestream
-
+          
           cd dist
           tar -czvf litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.tar.gz litestream
-          ../nfpm pkg --config nfpm.yml --packager deb --target litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.deb
+          ../nfpm pkg --config nfpm.yml --packager deb --target litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.deb 
 
       - name: Upload release tarball
         uses: actions/upload-release-asset@v1.0.2

.github/workflows/release.linux_static.yml

@@ -0,0 +1,62 @@
+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

@@ -0,0 +1,62 @@
+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

@@ -1,20 +0,0 @@
-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,16 +1,12 @@
-FROM golang:1.21.3 as builder
+FROM golang:1.16 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 []

LICENSE

@@ -199,4 +199,4 @@
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
-   limitations under the License.
+   limitations under the License.

Makefile

@@ -20,17 +20,11 @@ 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 disaster recovery tool for SQLite. It runs as a
+Litestream is a standalone streaming replication 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,28 +33,35 @@ 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.
+- Thanks to [Kurt Mackey](https://twitter.com/mrkurt) for feedback and testing. Also, thanks to fly.io for providing testing resources.
 - 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 Coleu](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 Coleuu](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.
 
-Litestream is now open to code contributions for bug fixes only. Features carry
+## Open-source, not open-contribution
-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.
 
-If you find mistakes in the documentation, please submit a fix to the
+[Similar to SQLite](https://www.sqlite.org/copyright.html), Litestream is open
-[documentation repository][docs].
+source but closed to code contributions. This keeps the code base free of
+proprietary or licensed code but it also helps me continue to maintain and build
+Litestream.
 
-[new-issue]: https://github.com/benbjohnson/litestream/issues/new
+As the author of [BoltDB](https://github.com/boltdb/bolt), I found that
+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,17 +87,15 @@ func (c *GenerationsCommand) Run(ctx context.Context, args []string) (err error)
 	for _, r := range replicas {
 		generations, err := r.Client.Generations(ctx)
 		if err != nil {
-			r.Logger().Error("cannot list generations", "error", err)
+			log.Printf("%s: cannot list generations: %s", r.Name(), 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 {
-				r.Logger().Error("cannot determine generation time bounds", "error", err)
+				log.Printf("%s: cannot determine generation time bounds: %s", r.Name(), err)
 				continue
 			}
 
@@ -124,7 +122,7 @@ cover.
 Usage:
 
 	litestream generations [arguments] DB_PATH
-
+	
 	litestream generations [arguments] REPLICA_URL
 
 Arguments:

cmd/litestream/main.go

@@ -5,7 +5,8 @@ import (
 	"errors"
 	"flag"
 	"fmt"
-	"log/slog"
+	"io/ioutil"
+	"log"
 	"net/url"
 	"os"
 	"os/user"
@@ -16,7 +17,6 @@ import (
 	"strings"
 	"time"
 
-	"filippo.io/age"
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/abs"
 	"github.com/benbjohnson/litestream/file"
@@ -36,11 +36,13 @@ 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 {
-		slog.Error("failed to run", "error", err)
+		log.Println(err)
 		os.Exit(1)
 	}
 }
@@ -83,26 +85,29 @@ 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(); err != nil {
+		if err := c.Run(ctx); err != nil {
 			return err
 		}
 
 		// Wait for signal to stop program.
 		select {
 		case err = <-c.execCh:
-			slog.Info("subprocess exited, litestream shutting down")
+			cancel()
+			fmt.Println("subprocess exited, litestream shutting down")
 		case sig := <-signalCh:
-			slog.Info("signal received, litestream shutting down")
+			cancel()
+			fmt.Println("signal received, litestream shutting down")
 
 			if c.cmd != nil {
-				slog.Info("sending signal to exec process")
+				fmt.Println("sending signal to exec process")
 				if err := c.cmd.Process.Signal(sig); err != nil {
 					return fmt.Errorf("cannot signal exec process: %w", err)
 				}
 
-				slog.Info("waiting for exec process to close")
+				fmt.Println("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)
 				}
@@ -113,7 +118,7 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 		if e := c.Close(); e != nil && err == nil {
 			err = e
 		}
-		slog.Info("litestream shut down")
+		fmt.Println("litestream shut down")
 		return err
 
 	case "restore":
@@ -169,16 +174,6 @@ 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.
@@ -222,7 +217,7 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 	}
 
 	// Read configuration.
-	buf, err := os.ReadFile(filename)
+	buf, err := ioutil.ReadFile(filename)
 	if os.IsNotExist(err) {
 		return config, fmt.Errorf("config file not found: %s", filename)
 	} else if err != nil {
@@ -248,43 +243,12 @@ 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"`
@@ -304,9 +268,6 @@ 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
 	}
@@ -362,12 +323,6 @@ 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.
@@ -394,22 +349,6 @@ 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 @@
-//go:build !windows
+// +build !windows
 
 package main
 

cmd/litestream/main_test.go

@@ -1,6 +1,7 @@
 package main_test
 
 import (
+	"io/ioutil"
 	"os"
 	"path/filepath"
 	"testing"
@@ -15,7 +16,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 := os.WriteFile(filename, []byte(`
+		if err := ioutil.WriteFile(filename, []byte(`
 access-key-id: XXX
 secret-access-key: YYY
 
@@ -47,7 +48,7 @@ dbs:
 		os.Setenv("LITESTREAM_TEST_1872363", "s3://foo/bar")
 
 		filename := filepath.Join(t.TempDir(), "litestream.yml")
-		if err := os.WriteFile(filename, []byte(`
+		if err := ioutil.WriteFile(filename, []byte(`
 dbs:
   - path: $LITESTREAM_TEST_0129380
     replicas:
@@ -74,7 +75,7 @@ dbs:
 		os.Setenv("LITESTREAM_TEST_9847533", "s3://foo/bar")
 
 		filename := filepath.Join(t.TempDir(), "litestream.yml")
-		if err := os.WriteFile(filename, []byte(`
+		if err := ioutil.WriteFile(filename, []byte(`
 dbs:
   - path: /path/to/db
     replicas:

cmd/litestream/main_windows.go

@@ -1,11 +1,11 @@
-//go:build windows
+// +build windows
 
 package main
 
 import (
 	"context"
 	"io"
-	"log/slog"
+	"log"
 	"os"
 	"os/signal"
 
@@ -36,16 +36,16 @@ func runWindowsService(ctx context.Context) error {
 	defer elog.Close()
 
 	// Set eventlog as log writer while running.
-	slog.SetDefault(slog.New(slog.NewTextHandler((*eventlogWriter)(elog), nil)))
+	log.SetOutput((*eventlogWriter)(elog))
-	defer slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, nil)))
+	defer log.SetOutput(os.Stderr)
 
-	slog.Info("Litestream service starting")
+	log.Print("Litestream service starting")
 
 	if err := svc.Run(serviceName, &windowsService{ctx: ctx}); err != nil {
 		return errStop
 	}
 
-	slog.Info("Litestream service stopped")
+	log.Print("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 {
-		slog.Error("cannot load configuration", "error", err)
+		log.Printf("cannot load configuration: %s", err)
 		return true, 1
 	}
 
 	// Execute replication command.
-	if err := c.Run(); err != nil {
+	if err := c.Run(s.ctx); err != nil {
-		slog.Error("cannot replicate", "error", err)
+		log.Printf("cannot replicate: %s", 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:
-				slog.Error("Litestream service received unexpected change request", "cmd", req.Cmd)
+				log.Printf("Litestream service received unexpected change request cmd: %d", req.Cmd)
 			}
 		}
 	}

cmd/litestream/replicate.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log/slog"
+	"log"
 	"net"
 	"net/http"
 	_ "net/http/pprof"
@@ -42,6 +42,7 @@ 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 {
@@ -79,17 +80,27 @@ 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() (err error) {
+func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 	// Display version information.
-	slog.Info("litestream", "version", Version)
+	log.Printf("litestream %s", Version)
 
 	// Setup databases.
 	if len(c.Config.DBs) == 0 {
-		slog.Error("no databases specified in configuration")
+		log.Println("no databases specified in configuration")
 	}
 
 	for _, dbConfig := range c.Config.DBs {
@@ -107,22 +118,21 @@ func (c *ReplicateCommand) Run() (err error) {
 
 	// Notify user that initialization is done.
 	for _, db := range c.DBs {
-		slog.Info("initialized db", "path", db.Path())
+		log.Printf("initialized db: %s", 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:
-				slog.Info("replicating to", "path", client.Path())
+				log.Printf("replicating to: name=%q type=%q path=%q", r.Name(), client.Type(), client.Path())
 			case *s3.ReplicaClient:
-				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "region", client.Region, "endpoint", client.Endpoint)
+				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)
 			case *gcs.ReplicaClient:
-				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path)
+				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)
 			case *abs.ReplicaClient:
-				slog.Info("replicating to", "bucket", client.Bucket, "path", client.Path, "endpoint", client.Endpoint)
+				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)
 			case *sftp.ReplicaClient:
-				slog.Info("replicating to", "host", client.Host, "user", client.User, "path", client.Path)
+				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)
 			default:
-				slog.Info("replicating to")
+				log.Printf("replicating to: name=%q type=%q", r.Name(), client.Type())
 			}
 		}
 	}
@@ -136,11 +146,11 @@ func (c *ReplicateCommand) Run() (err error) {
 			hostport = net.JoinHostPort("localhost", port)
 		}
 
-		slog.Info("serving metrics on", "url", fmt.Sprintf("http://%s/metrics", hostport))
+		log.Printf("serving metrics on http://%s/metrics", hostport)
 		go func() {
 			http.Handle("/metrics", promhttp.Handler())
 			if err := http.ListenAndServe(c.Config.Addr, nil); err != nil {
-				slog.Error("cannot start metrics server", "error", err)
+				log.Printf("cannot start metrics server: %s", err)
 			}
 		}()
 	}
@@ -152,7 +162,7 @@ func (c *ReplicateCommand) Run() (err error) {
 			return fmt.Errorf("cannot parse exec command: %w", err)
 		}
 
-		c.cmd = exec.Command(execArgs[0], execArgs[1:]...)
+		c.cmd = exec.CommandContext(ctx, execArgs[0], execArgs[1:]...)
 		c.cmd.Env = os.Environ()
 		c.cmd.Stdout = os.Stdout
 		c.cmd.Stderr = os.Stderr
@@ -168,8 +178,8 @@ func (c *ReplicateCommand) Run() (err error) {
 // Close closes all open databases.
 func (c *ReplicateCommand) Close() (err error) {
 	for _, db := range c.DBs {
-		if e := db.Close(context.Background()); e != nil {
+		if e := db.SoftClose(); e != nil {
-			db.Logger.Error("error closing db", "error", e)
+			log.Printf("error closing db: path=%s err=%s", db.Path(), e)
 			if err == nil {
 				err = e
 			}
@@ -181,7 +191,7 @@ func (c *ReplicateCommand) Close() (err error) {
 // Usage prints the help screen to STDOUT.
 func (c *ReplicateCommand) Usage() {
 	fmt.Printf(`
-The replicate command starts a server to monitor & replicate databases.
+The replicate command starts a server to monitor & replicate databases. 
 You can specify your database & replicas in a configuration file or you can
 replicate a single database file by specifying its path and its replicas in the
 command line arguments.
@@ -205,5 +215,8 @@ 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/slog"
+	"log"
 	"os"
 	"strconv"
 	"time"
@@ -19,6 +19,7 @@ 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)
@@ -30,6 +31,7 @@ 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
@@ -46,6 +48,11 @@ 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)) {
@@ -53,7 +60,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 {
-			slog.Info("database already exists, skipping")
+			fmt.Println("database already exists, skipping")
 			return nil
 		} else if err != nil {
 			return err
@@ -63,7 +70,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 {
-			slog.Info("database already exists, skipping")
+			fmt.Println("database already exists, skipping")
 			return nil
 		} else if err != nil {
 			return err
@@ -74,7 +81,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 {
-			slog.Info("no matching backups found")
+			fmt.Println("no matching backups found")
 			return nil
 		}
 		return fmt.Errorf("no matching backups found")
@@ -197,6 +204,9 @@ 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/slog"
+	"log"
 	"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 {
-			slog.Error("cannot determine snapshots", "error", err)
+			log.Printf("cannot determine snapshots: %s", err)
 			continue
 		}
 		for _, info := range infos {

cmd/litestream/wal.go

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

db.go

@@ -10,7 +10,8 @@ import (
 	"fmt"
 	"hash/crc64"
 	"io"
-	"log/slog"
+	"io/ioutil"
+	"log"
 	"math"
 	"math/rand"
 	"os"
@@ -30,7 +31,6 @@ const (
 	DefaultCheckpointInterval = 1 * time.Minute
 	DefaultMinCheckpointPageN = 1000
 	DefaultMaxCheckpointPageN = 10000
-	DefaultTruncatePageN      = 500000
 )
 
 // MaxIndex is the maximum possible WAL index.
@@ -44,13 +44,11 @@ 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
@@ -85,16 +83,6 @@ 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.
@@ -106,60 +94,18 @@ 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,
+		path:   path,
-		metaPath: filepath.Join(dir, "."+file+MetaDirSuffix),
+		notify: make(chan struct{}),
-		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)
@@ -196,24 +142,20 @@ func (db *DB) WALPath() string {
 
 // MetaPath returns the path to the database metadata.
 func (db *DB) MetaPath() string {
-	return db.metaPath
+	dir, file := filepath.Split(db.path)
-}
+	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.
@@ -240,7 +182,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) {
-	des, err := os.ReadDir(filepath.Join(db.GenerationPath(generation), "wal"))
+	fis, err := ioutil.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 {
@@ -248,14 +190,7 @@ func (db *DB) CurrentShadowWALIndex(generation string) (index int, size int64, e
 	}
 
 	// Find highest wal index.
-	for _, de := range des {
+	for _, fi := range fis {
-		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 {
@@ -343,7 +278,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)
 	}
 
@@ -356,12 +291,26 @@ func (db *DB) Open() (err error) {
 	return nil
 }
 
-// Close flushes outstanding WAL writes to replicas, releases the read lock,
+// Close releases the read lock & closes the database. This method should only
-// and closes the database. Takes a context for final sync.
+// be called by tests as it causes the underlying database to be checkpointed.
-func (db *DB) Close(ctx context.Context) (err error) {
+func (db *DB) Close() (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 {
@@ -376,7 +325,7 @@ func (db *DB) Close(ctx context.Context) (err error) {
 				err = e
 			}
 		}
-		r.Stop(true)
+		r.Stop(!soft)
 	}
 
 	// Release the read lock to allow other applications to handle checkpointing.
@@ -386,18 +335,14 @@ func (db *DB) Close(ctx context.Context) (err error) {
 		}
 	}
 
-	if db.db != nil {
+	// Only perform full close if this is not a soft close.
+	// 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
 }
 
@@ -447,9 +392,8 @@ func (db *DB) init() (err error) {
 	dsn := db.path
 	dsn += fmt.Sprintf("?_busy_timeout=%d", BusyTimeout.Milliseconds())
 
-	// Connect to SQLite database. Use the driver registered with a hook to
+	// Connect to SQLite database.
-	// prevent WAL files from being removed.
+	if db.db, err = sql.Open("sqlite3", dsn); err != nil {
-	if db.db, err = sql.Open("litestream-sqlite3", dsn); err != nil {
 		return err
 	}
 
@@ -509,13 +453,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 {
-		db.Logger.Warn("init: cannot determine last wal position, clearing generation", "error", err)
+		log.Printf("%s: init: cannot determine last wal position, clearing generation; %s", db.path, err)
 		if err := os.Remove(db.GenerationNamePath()); err != nil && !os.IsNotExist(err) {
 			return fmt.Errorf("remove generation name: %w", err)
 		}
@@ -585,8 +529,8 @@ func (db *DB) cleanGenerations() error {
 		return err
 	}
 
-	dir := filepath.Join(db.metaPath, "generations")
+	dir := filepath.Join(db.MetaPath(), "generations")
-	fis, err := os.ReadDir(dir)
+	fis, err := ioutil.ReadDir(dir)
 	if os.IsNotExist(err) {
 		return nil
 	} else if err != nil {
@@ -633,7 +577,7 @@ func (db *DB) cleanWAL() error {
 
 	// Remove all WAL files for the generation before the lowest index.
 	dir := db.ShadowWALDir(generation)
-	fis, err := os.ReadDir(dir)
+	fis, err := ioutil.ReadDir(dir)
 	if os.IsNotExist(err) {
 		return nil
 	} else if err != nil {
@@ -689,7 +633,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 := os.ReadFile(db.GenerationNamePath())
+	buf, err := ioutil.ReadFile(db.GenerationNamePath())
 	if os.IsNotExist(err) {
 		return "", nil
 	} else if err != nil {
@@ -715,7 +659,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
 	}
@@ -731,7 +675,7 @@ func (db *DB) createGeneration() (string, error) {
 	if db.fileInfo != nil {
 		mode = db.fileInfo.Mode()
 	}
-	if err := os.WriteFile(generationNamePath+".tmp", []byte(generation+"\n"), mode); err != nil {
+	if err := ioutil.WriteFile(generationNamePath+".tmp", []byte(generation+"\n"), mode); err != nil {
 		return "", fmt.Errorf("write generation temp file: %w", err)
 	}
 	uid, gid := internal.Fileinfo(db.fileInfo)
@@ -757,7 +701,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 	if err := db.init(); err != nil {
 		return err
 	} else if db.db == nil {
-		db.Logger.Debug("sync: no database found")
+		Tracef("%s: sync: no database found", db.path)
 		return nil
 	}
 
@@ -783,7 +727,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 	if err != nil {
 		return fmt.Errorf("cannot verify wal state: %w", err)
 	}
-	db.Logger.Debug("sync", "info", &info)
+	Tracef("%s: sync: info=%#v", db.path, info)
 
 	// Track if anything in the shadow WAL changes and then notify at the end.
 	changed := info.walSize != info.shadowWALSize || info.restart || info.reason != ""
@@ -794,7 +738,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)
 		}
-		db.Logger.Info("sync: new generation", "generation", info.generation, "reason", info.reason)
+		log.Printf("%s: sync: new generation %q, %s", db.path, info.generation, info.reason)
 
 		// Clear shadow wal info.
 		info.shadowWALPath = db.ShadowWALPath(info.generation, 0)
@@ -805,7 +749,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 	}
 
 	// Synchronize real WAL with current shadow WAL.
-	origWALSize, newWALSize, err := db.syncWAL(info)
+	newWALSize, err := db.syncWAL(info)
 	if err != nil {
 		return fmt.Errorf("sync wal: %w", err)
 	}
@@ -814,9 +758,7 @@ 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.TruncatePageN > 0 && origWALSize >= calcWALSize(db.pageSize, db.TruncatePageN) {
+	if db.MaxCheckpointPageN > 0 && newWALSize >= calcWALSize(db.pageSize, db.MaxCheckpointPageN) {
-		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
@@ -850,7 +792,7 @@ func (db *DB) Sync(ctx context.Context) (err error) {
 		db.notify = make(chan struct{})
 	}
 
-	db.Logger.Debug("sync: ok")
+	Tracef("%s: sync: ok", db.path)
 
 	return nil
 }
@@ -975,29 +917,29 @@ type syncInfo struct {
 }
 
 // syncWAL copies pending bytes from the real WAL to the shadow WAL.
-func (db *DB) syncWAL(info syncInfo) (origSize int64, newSize int64, err error) {
+func (db *DB) syncWAL(info syncInfo) (newSize int64, err error) {
 	// Copy WAL starting from end of shadow WAL. Exit if no new shadow WAL needed.
-	origSize, newSize, err = db.copyToShadowWAL(info.shadowWALPath)
+	newSize, err = db.copyToShadowWAL(info.shadowWALPath)
 	if err != nil {
-		return origSize, newSize, fmt.Errorf("cannot copy to shadow wal: %w", err)
+		return newSize, fmt.Errorf("cannot copy to shadow wal: %w", err)
 	} else if !info.restart {
-		return origSize, newSize, nil // If no restart required, exit.
+		return 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, 0, fmt.Errorf("cannot parse shadow wal filename: %s", base)
+		return 0, fmt.Errorf("cannot parse shadow wal filename: %s", base)
 	}
 
 	// Start a new shadow WAL file with next index.
 	newShadowWALPath := filepath.Join(dir, FormatWALPath(index+1))
 	newSize, err = db.initShadowWALFile(newShadowWALPath)
 	if err != nil {
-		return 0, 0, fmt.Errorf("cannot init shadow wal file: name=%s err=%w", newShadowWALPath, err)
+		return 0, fmt.Errorf("cannot init shadow wal file: name=%s err=%w", newShadowWALPath, err)
 	}
-	return origSize, newSize, nil
+	return newSize, nil
 }
 
 func (db *DB) initShadowWALFile(filename string) (int64, error) {
@@ -1026,103 +968,87 @@ 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 := os.WriteFile(filename, hdr, mode); err != nil {
+	} else if err := ioutil.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) (origWalSize int64, newSize int64, err error) {
+func (db *DB) copyToShadowWAL(filename string) (newSize int64, err error) {
-	logger := db.Logger.With("filename", filename)
+	Tracef("%s: copy-shadow: %s", db.path, filename)
-	logger.Debug("copy-shadow")
 
 	r, err := os.Open(db.WALPath())
 	if err != nil {
-		return 0, 0, err
+		return 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, 0, err
+		return 0, err
 	}
 	defer w.Close()
 
-	fi, err = w.Stat()
+	fi, err := w.Stat()
 	if err != nil {
-		return 0, 0, err
+		return 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, 0, fmt.Errorf("read header: %w", err)
+		return 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, 0, err
+		return 0, err
 	}
 
 	// Read previous checksum.
 	chksum0, chksum1, err := readLastChecksumFrom(w, db.pageSize)
 	if err != nil {
-		return 0, 0, fmt.Errorf("last checksum: %w", err)
+		return 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, 0, fmt.Errorf("real wal seek: %w", err)
+		return 0, fmt.Errorf("real wal seek: %w", err)
 	} else if _, err := w.Seek(origSize, io.SeekStart); err != nil {
-		return 0, 0, fmt.Errorf("shadow wal seek: %w", err)
+		return 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 {
-			logger.Debug("copy-shadow: break", "offset", offset, "error", err)
+			Tracef("%s: copy-shadow: break %s @ %d; err=%s", db.path, filename, offset, err)
 			break // end of file or partial page
 		} else if err != nil {
-			return 0, 0, fmt.Errorf("read wal: %w", err)
+			return 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 {
-			logger.Debug("copy-shadow: break: salt mismatch")
+			Tracef("%s: copy-shadow: break: salt mismatch", db.path)
 			break
 		}
 
@@ -1132,60 +1058,38 @@ func (db *DB) copyToShadowWAL(filename string) (origWalSize int64, newSize int64
 		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 {
-			logger.Debug("copy shadow: checksum mismatch, skipping", "offset", offset, "check", fmt.Sprintf("(%x,%x) != (%x,%x)", chksum0, chksum1, fchksum0, fchksum1))
+			Tracef("%s: copy shadow: checksum mismatch, skipping: offset=%d (%x,%x) != (%x,%x)", db.path, offset, chksum0, chksum1, fchksum0, fchksum1)
 			break
 		}
 
-		// Write page to temporary WAL file.
+		// Add page to the new size of the shadow WAL.
-		if _, err := f.Write(frame); err != nil {
+		buf.Write(frame)
-			return 0, 0, fmt.Errorf("write temp shadow wal: %w", err)
-		}
 
-		logger.Debug("copy-shadow: ok", "offset", offset, "salt", fmt.Sprintf("%x %x", salt0, salt1))
+		Tracef("%s: copy-shadow: ok %s offset=%d salt=%x %x", db.path, filename, offset, salt0, salt1)
 		offset += int64(len(frame))
 
-		// Update new size if written frame was a commit record.
+		// Flush to shadow WAL if 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
 		}
 	}
 
-	// If no WAL writes found, exit.
+	// Sync & close.
-	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, 0, err
+		return 0, err
 	} else if err := w.Close(); err != nil {
-		return 0, 0, err
+		return 0, err
 	}
 
 	// Track total number of bytes written to WAL.
-	db.totalWALBytesCounter.Add(float64(walByteN))
+	db.totalWALBytesCounter.Add(float64(lastCommitSize - origSize))
 
-	return origWalSize, lastCommitSize, nil
+	return lastCommitSize, nil
 }
 
 // ShadowWALReader opens a reader for a shadow WAL file at a given position.
@@ -1339,15 +1243,9 @@ func (db *DB) Checkpoint(ctx context.Context, mode string) (err error) {
 	return db.checkpoint(ctx, generation, mode)
 }
 
-// checkpoint performs a checkpoint on the WAL file and initializes a
+// checkpointAndInit 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
@@ -1360,7 +1258,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)
 	}
 
@@ -1395,7 +1293,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)
 	}
 
@@ -1454,7 +1352,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
 	}
-	db.Logger.Debug("checkpoint", "mode", mode, "result", fmt.Sprintf("%d,%d,%d", row[0], row[1], row[2]))
+	Tracef("%s: checkpoint: mode=%v (%d,%d,%d)", db.path, mode, row[0], row[1], row[2])
 
 	// Reacquire the read lock immediately after the checkpoint.
 	if err := db.acquireReadLock(); err != nil {
@@ -1479,7 +1377,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) {
-			db.Logger.Error("sync error", "error", err)
+			log.Printf("%s: sync error: %s", db.path, err)
 		}
 	}
 }
@@ -1583,19 +1481,6 @@ 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
 
@@ -1623,6 +1508,10 @@ 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,6 +3,7 @@ package litestream_test
 import (
 	"context"
 	"database/sql"
+	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strings"
@@ -263,14 +264,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(context.Background()); err != nil {
+		} else if err := db.Close(); err != nil {
 			t.Fatal(err)
 		} else if err := sqldb.Close(); err != nil {
 			t.Fatal(err)
 		}
 
-		// Remove WAL file.
+		// Verify WAL does not exist.
-		if err := os.Remove(db.WALPath()); err != nil {
+		if _, err := os.Stat(db.WALPath()); !os.IsNotExist(err) {
 			t.Fatal(err)
 		}
 
@@ -313,7 +314,7 @@ func TestDB_Sync(t *testing.T) {
 		}
 
 		// Fully close which should close WAL file.
-		if err := db.Close(context.Background()); err != nil {
+		if err := db.Close(); err != nil {
 			t.Fatal(err)
 		} else if err := sqldb.Close(); err != nil {
 			t.Fatal(err)
@@ -366,16 +367,16 @@ func TestDB_Sync(t *testing.T) {
 		pos0, err := db.Pos()
 		if err != nil {
 			t.Fatal(err)
-		} else if err := db.Close(context.Background()); err != nil {
+		} else if err := db.Close(); 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 := os.ReadFile(shadowWALPath); err != nil {
+		if buf, err := ioutil.ReadFile(shadowWALPath); err != nil {
 			t.Fatal(err)
-		} else if err := os.WriteFile(shadowWALPath, append(buf[:litestream.WALHeaderSize-8], 0, 0, 0, 0, 0, 0, 0, 0), 0600); err != nil {
+		} else if err := ioutil.WriteFile(shadowWALPath, append(buf[:litestream.WALHeaderSize-8], 0, 0, 0, 0, 0, 0, 0, 0), 0600); err != nil {
 			t.Fatal(err)
 		}
 
@@ -412,7 +413,7 @@ func TestDB_Sync(t *testing.T) {
 		}
 
 		// Close & truncate shadow WAL to simulate a partial header write.
-		if err := db.Close(context.Background()); err != nil {
+		if err := db.Close(); err != nil {
 			t.Fatal(err)
 		} else if err := os.Truncate(db.ShadowWALPath(pos0.Generation, pos0.Index), litestream.WALHeaderSize-1); err != nil {
 			t.Fatal(err)
@@ -457,7 +458,7 @@ func TestDB_Sync(t *testing.T) {
 		}
 
 		// Close & truncate shadow WAL to simulate a partial frame write.
-		if err := db.Close(context.Background()); err != nil {
+		if err := db.Close(); err != nil {
 			t.Fatal(err)
 		} else if err := os.Truncate(db.ShadowWALPath(pos0.Generation, pos0.Index), fi.Size()-1); err != nil {
 			t.Fatal(err)
@@ -503,7 +504,7 @@ func TestDB_Sync(t *testing.T) {
 		}
 
 		// Close & delete shadow WAL to simulate dir created but not WAL.
-		if err := db.Close(context.Background()); err != nil {
+		if err := db.Close(); err != nil {
 			t.Fatal(err)
 		} else if err := os.Remove(db.ShadowWALPath(pos0.Generation, pos0.Index)); err != nil {
 			t.Fatal(err)
@@ -625,7 +626,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(context.Background()); err != nil && !strings.Contains(err.Error(), `database is closed`) {
+	if err := db.Close(); 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.wxs

@@ -26,9 +26,9 @@
 			Description="Litestream $(var.Version) installer"
 			Compressed="yes"
 		/>
-
+		
 		<Media Id="1" Cabinet="litestream.cab" EmbedCab="yes"/>
-
+		
 		<MajorUpgrade
 			Schedule="afterInstallInitialize"
 			DowngradeErrorMessage="A later version of [ProductName] is already installed. Setup will now exit."

etc/litestream.yml

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

etc/s3_mock.py

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

go.mod

@@ -1,65 +1,21 @@
 module github.com/benbjohnson/litestream
 
-go 1.21
+go 1.16
 
 require (
-	cloud.google.com/go/storage v1.36.0
+	cloud.google.com/go/storage v1.15.0
-	filippo.io/age v1.1.1
+	github.com/Azure/azure-storage-blob-go v0.13.0 // indirect
-	github.com/Azure/azure-storage-blob-go v0.15.0
+	github.com/Azure/go-autorest/autorest v0.9.0 // indirect
-	github.com/aws/aws-sdk-go v1.49.5
+	github.com/aws/aws-sdk-go v1.27.0
-	github.com/mattn/go-shellwords v1.0.12
+	github.com/davecgh/go-spew v1.1.1
-	github.com/mattn/go-sqlite3 v1.14.19
+	github.com/mattn/go-shellwords v1.0.11 // indirect
-	github.com/pierrec/lz4/v4 v4.1.19
+	github.com/mattn/go-sqlite3 v1.14.5
-	github.com/pkg/sftp v1.13.6
+	github.com/pierrec/lz4/v4 v4.1.3
-	github.com/prometheus/client_golang v1.17.0
+	github.com/pkg/sftp v1.13.0 // indirect
-	golang.org/x/crypto v0.17.0
+	github.com/prometheus/client_golang v1.9.0
-	golang.org/x/sync v0.5.0
+	golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a // indirect
-	golang.org/x/sys v0.15.0
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
-	google.golang.org/api v0.154.0
+	golang.org/x/sys v0.0.0-20210412220455-f1c623a9e750
+	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,4 +1,3 @@
-//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,4 +1,3 @@
-//go:build windows
 // +build windows
 
 package internal

litestream.go

@@ -13,8 +13,6 @@ import (
 	"strconv"
 	"strings"
 	"time"
-
-	"github.com/mattn/go-sqlite3"
 )
 
 // Naming constants.
@@ -44,25 +42,6 @@ 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
@@ -228,13 +207,11 @@ 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 i := range a {
+	for _, snapshot := range a {
-		snapshot := &a[i]
-
 		if snapshot.Generation != generation {
 			continue
 		} else if min == nil || snapshot.Index < min.Index {
-			min = snapshot
+			min = &snapshot
 		}
 	}
 	return min
@@ -540,6 +517,9 @@ 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,16 +128,6 @@ 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,7 +6,8 @@ import (
 	"fmt"
 	"hash/crc64"
 	"io"
-	"log/slog"
+	"io/ioutil"
+	"log"
 	"math"
 	"os"
 	"path/filepath"
@@ -14,7 +15,6 @@ import (
 	"sync"
 	"time"
 
-	"filippo.io/age"
 	"github.com/benbjohnson/litestream/internal"
 	"github.com/pierrec/lz4/v4"
 	"github.com/prometheus/client_golang/prometheus"
@@ -67,10 +67,6 @@ 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 {
@@ -96,15 +92,6 @@ 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 }
 
@@ -170,13 +157,13 @@ func (r *Replica) Sync(ctx context.Context) (err error) {
 	}
 	generation := dpos.Generation
 
-	r.Logger().Debug("replica sync", "position", dpos.String())
+	Tracef("%s(%s): replica sync: db.pos=%s", r.db.Path(), r.Name(), dpos)
 
 	// 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(ctx, generation)
+		snapshotN, err := r.snapshotN(generation)
 		if err != nil {
 			return err
 		} else if snapshotN == 0 {
@@ -192,7 +179,7 @@ func (r *Replica) Sync(ctx context.Context) (err error) {
 			return fmt.Errorf("cannot determine replica position: %s", err)
 		}
 
-		r.Logger().Debug("replica sync: calc new pos", "position", pos.String())
+		Tracef("%s(%s): replica sync: calc new pos: %s", r.db.Path(), r.Name(), pos)
 		r.mu.Lock()
 		r.pos = pos
 		r.mu.Unlock()
@@ -226,40 +213,16 @@ 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(ew)
+	zw := lz4.NewWriter(pw)
 
 	// Track total WAL bytes written to replica client.
 	walBytesCounter := replicaWALBytesCounterVec.WithLabelValues(r.db.Path(), r.Name())
@@ -279,7 +242,6 @@ func (r *Replica) syncWAL(ctx context.Context) (err error) {
 			return err
 		}
 		walBytesCounter.Add(float64(n))
-		bytesWritten += n
 	}
 
 	// Copy frames.
@@ -306,14 +268,11 @@ func (r *Replica) syncWAL(ctx context.Context) (err error) {
 			return err
 		}
 		walBytesCounter.Add(float64(n))
-		bytesWritten += n
 	}
 
-	// Flush LZ4 writer, encryption writer and close pipe.
+	// Flush LZ4 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
 	}
@@ -332,13 +291,12 @@ 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(ctx context.Context, generation string) (int, error) {
+func (r *Replica) snapshotN(generation string) (int, error) {
-	itr, err := r.Client.Snapshots(ctx, generation)
+	itr, err := r.Client.Snapshots(context.Background(), generation)
 	if err != nil {
 		return 0, err
 	}
@@ -376,16 +334,7 @@ func (r *Replica) calcPos(ctx context.Context, generation string) (pos Pos, err
 	}
 	defer rd.Close()
 
-	if len(r.AgeIdentities) > 0 {
+	n, err := io.Copy(ioutil.Discard, lz4.NewReader(rd))
-		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
 	}
@@ -483,15 +432,11 @@ 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.Checkpoint(ctx, CheckpointModePassive); err != nil {
+	if _, err := r.db.db.ExecContext(ctx, `PRAGMA wal_checkpoint(PASSIVE);`); err != nil {
 		return info, fmt.Errorf("pre-snapshot checkpoint: %w", err)
 	}
 
-	// Prevent internal checkpoints during snapshot.
+	// Acquire a read lock on the database during snapshot to prevent checkpoints.
-	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
@@ -525,23 +470,7 @@ 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 {
-		// We need to ensure the pipe is closed.
+		zr := lz4.NewWriter(pw)
-		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 {
@@ -551,13 +480,9 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
 			pw.CloseWithError(err)
 			return err
 		}
-		return wc.Close()
+		return pw.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
@@ -565,7 +490,8 @@ func (r *Replica) Snapshot(ctx context.Context) (info SnapshotInfo, err error) {
 		return info, err
 	}
 
-	logger.Info("snapshot written", "position", pos.String(), "elapsed", time.Since(startTime).String(), "sz", info.Size)
+	log.Printf("%s(%s): snapshot written %s/%08x", r.db.Path(), r.Name(), pos.Generation, pos.Index)
+
 	return info, nil
 }
 
@@ -632,7 +558,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)
 		}
-		r.Logger().Info("snapshot deleted", "generation", generation, "index", index)
+		log.Printf("%s(%s): snapshot deleted %s/%08x", r.db.Path(), r.Name(), generation, index)
 	}
 
 	return itr.Close()
@@ -664,8 +590,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
 }
 
@@ -701,7 +627,7 @@ func (r *Replica) monitor(ctx context.Context) {
 
 		// Synchronize the shadow wal into the replication directory.
 		if err := r.Sync(ctx); err != nil {
-			r.Logger().Error("monitor error", "error", err)
+			log.Printf("%s(%s): monitor error: %s", r.db.Path(), r.Name(), err)
 			continue
 		}
 	}
@@ -729,7 +655,7 @@ func (r *Replica) retainer(ctx context.Context) {
 			return
 		case <-ticker.C:
 			if err := r.EnforceRetention(ctx); err != nil {
-				r.Logger().Error("retainer error", "error", err)
+				log.Printf("%s(%s): retainer error: %s", r.db.Path(), r.Name(), err)
 				continue
 			}
 		}
@@ -742,31 +668,6 @@ 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()
 
@@ -776,7 +677,7 @@ func (r *Replica) snapshotter(ctx context.Context) {
 			return
 		case <-ticker.C:
 			if _, err := r.Snapshot(ctx); err != nil && err != ErrNoGeneration {
-				r.Logger().Error("snapshotter error", "error", err)
+				log.Printf("%s(%s): snapshotter error: %s", r.db.Path(), r.Name(), err)
 				continue
 			}
 		}
@@ -804,7 +705,7 @@ func (r *Replica) validator(ctx context.Context) {
 			return
 		case <-ticker.C:
 			if err := r.Validate(ctx); err != nil {
-				r.Logger().Error("validation error", "error", err)
+				log.Printf("%s(%s): validation error: %s", r.db.Path(), r.Name(), err)
 				continue
 			}
 		}
@@ -817,7 +718,7 @@ func (r *Replica) Validate(ctx context.Context) error {
 	db := r.DB()
 
 	// Restore replica to a temporary directory.
-	tmpdir, err := os.MkdirTemp("", "*-litestream")
+	tmpdir, err := ioutil.TempDir("", "*-litestream")
 	if err != nil {
 		return err
 	}
@@ -841,6 +742,7 @@ 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)
 	}
@@ -865,7 +767,7 @@ func (r *Replica) Validate(ctx context.Context) error {
 	if mismatch {
 		status = "mismatch"
 	}
-	r.Logger().Info("validator", "status", status, "db", fmt.Sprintf("%016x", chksum0), "replica", fmt.Sprintf("%016x", chksum1), "position", pos.String())
+	log.Printf("%s(%s): validator: status=%s db=%016x replica=%016x pos=%s", db.Path(), r.Name(), status, chksum0, chksum1, pos)
 
 	// Validate checksums match.
 	if mismatch {
@@ -883,6 +785,8 @@ 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()
 
@@ -905,7 +809,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() {
-			r.Logger().Info("validator: no replica position available")
+			log.Printf("%s(%s): validator: no replica position available", db.Path(), r.Name())
 			continue
 		}
 
@@ -961,7 +865,6 @@ 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) {
@@ -970,12 +873,6 @@ 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
@@ -990,9 +887,6 @@ 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
 		}
@@ -1067,6 +961,17 @@ 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)
@@ -1087,7 +992,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, minWALIndex, opt.Index, opt.Timestamp)
+	walSegmentMap, err := r.walSegmentMap(ctx, opt.Generation, opt.Index, opt.Timestamp)
 	if err != nil {
 		return fmt.Errorf("cannot find max wal index for restore: %w", err)
 	}
@@ -1101,7 +1006,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 != maxWALIndex {
+	if opt.Index != math.MaxInt32 && opt.Index != opt.Index {
 		return fmt.Errorf("unable to locate index %d in generation %q, highest index was %d", opt.Index, opt.Generation, maxWALIndex)
 	}
 
@@ -1113,19 +1018,19 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 	tmpPath := opt.OutputPath + ".tmp"
 
 	// Copy snapshot to output path.
-	r.Logger().Info("restoring snapshot", "generation", opt.Generation, "index", minWALIndex, "path", tmpPath)
+	logger.Printf("%s: restoring snapshot %s/%08x to %s", logPrefix, opt.Generation, minWALIndex, 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 {
-		r.Logger().Info("snapshot only, finalizing database")
+		logger.Printf("%s: snapshot only, finalizing database", logPrefix)
 		return os.Rename(tmpPath, opt.OutputPath)
 	}
 
 	// Begin processing WAL files.
-	r.Logger().Info("restoring wal files", "generation", opt.Generation, "index_min", minWALIndex, "index_max", maxWALIndex)
+	logger.Printf("%s: restoring wal files: generation=%s index=[%08x,%08x]", logPrefix, opt.Generation, minWALIndex, maxWALIndex)
 
 	// Fill input channel with all WAL indexes to be loaded in order.
 	// Verify every index has at least one offset.
@@ -1181,9 +1086,9 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 						return err
 					}
 
-					r.Logger().Info("downloaded wal",
+					logger.Printf("%s: downloaded wal %s/%08x elapsed=%s",
-						"generation", opt.Generation, "index", index,
+						logPrefix, opt.Generation, index,
-						"elapsed", time.Since(startTime).String(),
+						time.Since(startTime).String(),
 					)
 				}
 			}
@@ -1210,7 +1115,10 @@ 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)
 		}
-		r.Logger().Info("applied wal", "generation", opt.Generation, "index", index, "elapsed", time.Since(startTime).String())
+		logger.Printf("%s: applied wal %s/%08x elapsed=%s",
+			logPrefix, opt.Generation, index,
+			time.Since(startTime).String(),
+		)
 	}
 
 	// Ensure all goroutines finish. All errors should have been handled during
@@ -1220,7 +1128,7 @@ func (r *Replica) Restore(ctx context.Context, opt RestoreOptions) (err error) {
 	}
 
 	// Copy file to final location.
-	r.Logger().Info("renaming database from temporary location")
+	logger.Printf("%s: renaming database from temporary location", logPrefix)
 	if err := os.Rename(tmpPath, opt.OutputPath); err != nil {
 		return err
 	}
@@ -1281,7 +1189,7 @@ func (r *Replica) SnapshotIndexByIndex(ctx context.Context, generation string, i
 		}
 
 		// Use snapshot if it newer.
-		if snapshotIndex == -1 || snapshot.Index >= snapshotIndex {
+		if snapshotIndex == -1 || snapshotIndex >= snapshotIndex {
 			snapshotIndex = snapshot.Index
 		}
 	}
@@ -1295,29 +1203,22 @@ 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, minIndex, maxIndex int, maxTimestamp time.Time) (map[int][]int64, error) {
+func (r *Replica) walSegmentMap(ctx context.Context, generation string, 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()
 
-	a := []WALSegmentInfo{}
-	for itr.Next() {
-		a = append(a, itr.WALSegment())
-	}
-
-	sort.Sort(WALSegmentInfoSlice(a))
-
 	m := make(map[int][]int64)
-	for _, info := range a {
+	for itr.Next() {
+		info := itr.WALSegment()
+
 		// 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.
@@ -1358,15 +1259,6 @@ 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 {
@@ -1393,16 +1285,6 @@ 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))
 	}
 

replica_client.go

@@ -10,13 +10,13 @@ type ReplicaClient interface {
 	// Returns the type of client.
 	Type() string
 
-	// Returns a list of available generations. Order is undefined.
+	// Returns a list of available generations.
 	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. Order is undefined.
+	// Returns an iterator of all snapshots within a generation on the replica.
 	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. Order is undefined.
+	// Returns an iterator of all WAL segments within a generation on the replica.
 	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"
+	"io/ioutil"
 	"math/rand"
 	"os"
 	"path"
@@ -12,6 +12,7 @@ import (
 	"sort"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/abs"
@@ -21,6 +22,10 @@ import (
 	"github.com/benbjohnson/litestream/sftp"
 )
 
+func init() {
+	rand.Seed(time.Now().UnixNano())
+}
+
 var (
 	// Enables integration tests.
 	integration = flag.String("integration", "file", "")
@@ -75,14 +80,10 @@ func TestReplicaClient_Generations(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		// Fetch and sort generations.
+		// Verify returned generations.
-		got, err := c.Generations(context.Background())
+		if got, err := c.Generations(context.Background()); err != nil {
-		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)
 		}
 	})
@@ -192,7 +193,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 := io.ReadAll(r); err != nil {
+		} else if buf, err := ioutil.ReadAll(r); err != nil {
 			t.Fatal(err)
 		} else if err := r.Close(); err != nil {
 			t.Fatal(err)
@@ -223,7 +224,7 @@ func TestReplicaClient_SnapshotReader(t *testing.T) {
 		}
 		defer r.Close()
 
-		if buf, err := io.ReadAll(r); err != nil {
+		if buf, err := ioutil.ReadAll(r); err != nil {
 			t.Fatal(err)
 		} else if got, want := string(buf), "foo"; got != want {
 			t.Fatalf("ReadAll=%v, want %v", got, want)
@@ -377,7 +378,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 := io.ReadAll(r); err != nil {
+		} else if buf, err := ioutil.ReadAll(r); err != nil {
 			t.Fatal(err)
 		} else if err := r.Close(); err != nil {
 			t.Fatal(err)
@@ -408,7 +409,7 @@ func TestReplicaClient_WALReader(t *testing.T) {
 		}
 		defer r.Close()
 
-		if buf, err := io.ReadAll(r); err != nil {
+		if buf, err := ioutil.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,13 +13,6 @@ 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 {
@@ -39,6 +32,11 @@ 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)
@@ -68,47 +66,10 @@ 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(), dpos.Truncate()); err != nil {
+	} else if r, err := c.WALSegmentReader(context.Background(), litestream.Pos{Generation: generations[0], Index: 0, Offset: 0}); err != nil {
 		t.Fatal(err)
 	} else if b1, err := io.ReadAll(lz4.NewReader(r)); err != nil {
 		t.Fatal(err)
@@ -142,7 +103,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(), nextIndex(pos0); got != want {
+	} else if got, want := info.Pos(), pos0.Truncate(); got != want {
 		t.Fatalf("pos=%s, want %s", got, want)
 	}
 
@@ -166,20 +127,18 @@ 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(), nextIndex(pos1); got != want {
+	} else if got, want := info.Pos(), pos1.Truncate(); got != want {
 		t.Fatalf("pos=%v, want %v", got, want)
 	}
 
-	// Verify three snapshots exist.
+	// Verify two snapshots exist.
 	if infos, err := r.Snapshots(context.Background()); err != nil {
 		t.Fatal(err)
-	} else if got, want := len(infos), 3; got != want {
+	} else if got, want := len(infos), 2; 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(), nextIndex(pos0); got != want {
+	} else if got, want := infos[1].Pos(), pos1.Truncate(); 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,6 +16,7 @@ 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"
@@ -92,7 +93,6 @@ 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,8 +105,7 @@ 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 := &aws.Config{}
+	config := defaults.Get().Config
-
 	if c.AccessKeyID != "" || c.SecretAccessKey != "" {
 		config.Credentials = credentials.NewStaticCredentials(c.AccessKeyID, c.SecretAccessKey, "")
 	}
@@ -136,12 +135,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).HeadBucketWithContext(ctx, &s3.HeadBucketInput{
+	if out, err := s3.New(sess).GetBucketLocation(&s3.GetBucketLocationInput{
 		Bucket: aws.String(bucket),
 	}); err != nil {
 		return "", err
-	} else if out.BucketRegion != nil {
+	} else if out.LocationConstraint != nil {
-		return *out.BucketRegion, nil
+		return *out.LocationConstraint, nil
 	}
 	return DefaultRegion, nil
 }
@@ -161,7 +160,7 @@ func (c *ReplicaClient) Generations(ctx context.Context) ([]string, error) {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
 		for _, prefix := range page.CommonPrefixes {
-			name := path.Base(aws.StringValue(prefix.Prefix))
+			name := path.Base(*prefix.Prefix)
 			if !litestream.IsGenerationName(name) {
 				continue
 			}
@@ -209,14 +208,10 @@ func (c *ReplicaClient) DeleteGeneration(ctx context.Context, generation string)
 			n = len(objIDs)
 		}
 
-		out, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
+		if _, 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()
@@ -292,7 +287,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(aws.Int64Value(out.ContentLength)))
+	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "GET").Add(float64(*out.ContentLength))
 
 	return out.Body, nil
 }
@@ -308,14 +303,10 @@ func (c *ReplicaClient) DeleteSnapshot(ctx context.Context, generation string, i
 		return fmt.Errorf("cannot determine snapshot path: %w", err)
 	}
 
-	out, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
+	if _, 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
 	}
 
@@ -386,7 +377,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(aws.Int64Value(out.ContentLength)))
+	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "GET").Add(float64(*out.ContentLength))
 
 	return out.Body, nil
 }
@@ -414,16 +405,13 @@ func (c *ReplicaClient) DeleteWALSegments(ctx context.Context, a []litestream.Po
 		}
 
 		// Delete S3 objects in bulk.
-		out, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
+		if _, 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:]
@@ -466,14 +454,10 @@ func (c *ReplicaClient) DeleteAll(ctx context.Context) error {
 			n = len(objIDs)
 		}
 
-		out, err := c.s3.DeleteObjectsWithContext(ctx, &s3.DeleteObjectsInput{
+		if _, 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()
@@ -527,7 +511,7 @@ func (itr *snapshotIterator) fetch() error {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
 		for _, obj := range page.Contents {
-			key := path.Base(aws.StringValue(obj.Key))
+			key := path.Base(*obj.Key)
 			index, err := litestream.ParseSnapshotPath(key)
 			if err != nil {
 				continue
@@ -536,7 +520,7 @@ func (itr *snapshotIterator) fetch() error {
 			info := litestream.SnapshotInfo{
 				Generation: itr.generation,
 				Index:      index,
-				Size:       aws.Int64Value(obj.Size),
+				Size:       *obj.Size,
 				CreatedAt:  obj.LastModified.UTC(),
 			}
 
@@ -630,7 +614,7 @@ func (itr *walSegmentIterator) fetch() error {
 		internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "LIST").Inc()
 
 		for _, obj := range page.Contents {
-			key := path.Base(aws.StringValue(obj.Key))
+			key := path.Base(*obj.Key)
 			index, offset, err := litestream.ParseWALSegmentPath(key)
 			if err != nil {
 				continue
@@ -640,7 +624,7 @@ func (itr *walSegmentIterator) fetch() error {
 				Generation: itr.generation,
 				Index:      index,
 				Offset:     offset,
-				Size:       aws.Int64Value(obj.Size),
+				Size:       *obj.Size,
 				CreatedAt:  obj.LastModified.UTC(),
 			}
 
@@ -716,9 +700,6 @@ 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)
@@ -745,7 +726,6 @@ 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$`)
 )
 
@@ -757,15 +737,3 @@ 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,6 +8,7 @@ import (
 	"net"
 	"os"
 	"path"
+	"sort"
 	"sync"
 	"time"
 
@@ -140,6 +141,8 @@ func (c *ReplicaClient) Generations(ctx context.Context) (_ []string, err error)
 		generations = append(generations, name)
 	}
 
+	sort.Strings(generations)
+
 	return generations, nil
 }
 
@@ -226,6 +229,8 @@ func (c *ReplicaClient) Snapshots(ctx context.Context, generation string) (_ lit
 		})
 	}
 
+	sort.Sort(litestream.SnapshotInfoSlice(infos))
+
 	return litestream.NewSnapshotInfoSliceIterator(infos), nil
 }
 
@@ -358,6 +363,8 @@ func (c *ReplicaClient) WALSegments(ctx context.Context, generation string) (_ l
 		})
 	}
 
+	sort.Sort(litestream.WALSegmentInfoSlice(infos))
+
 	return litestream.NewWALSegmentInfoSliceIterator(infos), nil
 }