Document -addr flag on replicate command

This commit adds an http server and client for streaming snapshots
and WAL pages from an upstream Litestream primary to a read-only
replica.

.github/CONTRIBUTING.md

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

.github/pull_request_template.md

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

.github/workflows/build_and_test.yml

@@ -0,0 +1,25 @@
+name: "Build and Unit Test"
+on: pull_request
+
+jobs:
+  build:
+    name: Build
+    runs-on: ubuntu-18.04
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: actions/setup-go@v2
+        with:
+          go-version: '1.17'
+
+      - uses: actions/cache@v2
+        with:
+          path: ~/go/pkg/mod
+          key: ${{ inputs.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: ${{ inputs.os }}-go-
+
+      - name: Build binary
+        run: go install ./cmd/litestream
+
+      - name: Run unit tests
+        run: make testdata && go test -v ./...

.github/workflows/codeql-analysis.yml

@@ -0,0 +1,38 @@
+name: "CodeQL"
+
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+  schedule:
+    - cron: '20 16 * * 4'
+
+jobs:
+  analyze:
+    name: Analyze
+    runs-on: ubuntu-latest
+    permissions:
+      actions: read
+      contents: read
+      security-events: write
+
+    strategy:
+      fail-fast: false
+      matrix:
+        language: [ 'go' ]
+
+    steps:
+    - name: Checkout repository
+      uses: actions/checkout@v2
+
+    - name: Initialize CodeQL
+      uses: github/codeql-action/init@v1
+      with:
+        languages: ${{ matrix.language }}
+
+    - name: Autobuild
+      uses: github/codeql-action/autobuild@v1
+
+    - name: Perform CodeQL Analysis
+      uses: github/codeql-action/analyze@v1

.github/workflows/golangci-lint.yml

@@ -0,0 +1,18 @@
+name: golangci-lint
+on:
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  golangci:
+    name: lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      
+      - uses: golangci/golangci-lint-action@v2
+        with:
+          version: latest
+          args: --timeout=10m

.github/workflows/integration_test.yml

@@ -0,0 +1,138 @@
+name: Integration Tests
+on: 
+  pull_request:
+    branches-ignore:
+      - "dependabot/**"
+
+jobs:
+  s3-integration-test:
+    name: Run S3 Integration Tests
+    runs-on: ubuntu-18.04
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: actions/setup-go@v2
+        with:
+          go-version: '1.17'
+
+      - 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=TestReplicaClient ./integration -replica-type 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-18.04
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: actions/setup-go@v2
+        with:
+          go-version: '1.17'
+
+      - uses: actions/cache@v2
+        with:
+          path: ~/go/pkg/mod
+          key: ${{ inputs.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: ${{ inputs.os }}-go-
+
+      - name: Extract GCP credentials
+        run: 'echo "$GOOGLE_APPLICATION_CREDENTIALS" > /opt/gcp.json'
+        shell: bash
+        env:
+          GOOGLE_APPLICATION_CREDENTIALS: ${{secrets.GOOGLE_APPLICATION_CREDENTIALS}}
+
+      - run: go test -v -run=TestReplicaClient ./integration -replica-type gs
+        env:
+          GOOGLE_APPLICATION_CREDENTIALS:  /opt/gcp.json
+          LITESTREAM_GS_BUCKET:            integration.litestream.io
+
+  abs-integration-test:
+    name: Run Azure Blob Store Integration Tests
+    runs-on: ubuntu-18.04
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: actions/setup-go@v2
+        with:
+          go-version: '1.17'
+
+      - uses: actions/cache@v2
+        with:
+          path: ~/go/pkg/mod
+          key: ${{ inputs.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: ${{ inputs.os }}-go-
+
+      - run: go test -v -run=TestReplicaClient ./integration -replica-type 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-18.04
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: actions/setup-go@v2
+        with:
+          go-version: '1.17'
+
+      - uses: actions/cache@v2
+        with:
+          path: ~/go/pkg/mod
+          key: ${{ inputs.os }}-go-${{ hashFiles('**/go.sum') }}
+          restore-keys: ${{ inputs.os }}-go-
+
+      - name: Extract SSH key
+        run: 'echo "$LITESTREAM_SFTP_KEY" > /opt/id_ed25519'
+        shell: bash
+        env:
+          LITESTREAM_SFTP_KEY: ${{secrets.LITESTREAM_SFTP_KEY}}
+
+      - name: Run sftp tests w/ key
+        run: go test -v -run=TestReplicaClient ./integration -replica-type sftp
+        env:
+          LITESTREAM_SFTP_HOST: litestream-test-sftp.fly.dev:2222
+          LITESTREAM_SFTP_USER: litestream
+          LITESTREAM_SFTP_PATH: /litestream
+          LITESTREAM_SFTP_KEY_PATH:      /opt/id_ed25519
+
+      - name: Run sftp tests w/ password
+        run: go test -v -run=TestReplicaClient ./integration -replica-type sftp
+        env:
+          LITESTREAM_SFTP_HOST:     litestream-test-sftp.fly.dev:2222
+          LITESTREAM_SFTP_USER:     litestream
+          LITESTREAM_SFTP_PASSWORD: ${{ secrets.LITESTREAM_SFTP_PASSWORD }}
+          LITESTREAM_SFTP_PATH:     /litestream
+
+  long-running-test:
+    name: Run Long-Running Test
+    runs-on: ubuntu-18.04
+    steps:
+      - uses: actions/checkout@v2
+
+      - uses: actions/setup-go@v2
+        with: 
+          go-version: '1.17'
+
+      - 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

.github/workflows/release.docker.yml

@@ -0,0 +1,51 @@
+on:
+  release:
+    types:
+      - published
+  pull_request:
+    types:
+      - opened
+      - synchronize
+      - reopened
+    branches-ignore:
+      - "dependabot/**"
+
+name: Release (Docker)
+jobs:
+  docker:
+    runs-on: ubuntu-latest
+    env:
+      PLATFORMS: "${{ github.event_name == 'release' && 'linux/amd64,linux/arm64,linux/arm/v7' || 'linux/amd64' }}"
+      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

@@ -1,42 +1,72 @@
 on:
   release:
     types:
-      - created
+      - published
+  pull_request:
+    types:
+      - opened
+      - synchronize
+      - reopened
+    branches-ignore:
+      - "dependabot/**"
 
-name: release (linux)
+name: Release (Linux)
 jobs:
   build:
     runs-on: ubuntu-18.04
     strategy:
       matrix:
         include:
-          - arch: amd64
+          - arch:   amd64
-            cc:   gcc
+            cc:     gcc
+
+          - arch:   amd64
+            cc:     gcc
+            static: true
+            deploy_test_runner: true
+
           - arch: arm64
             cc:   aarch64-linux-gnu-gcc
+
+          - arch: arm64
+            cc:   aarch64-linux-gnu-gcc
+            static: true
+
           - arch: arm
             arm:  6
             cc:   arm-linux-gnueabi-gcc
+
+          - arch: arm
+            arm:  6
+            cc:   arm-linux-gnueabi-gcc
+            static: true
+
           - arch: arm
             arm:  7
             cc:   arm-linux-gnueabihf-gcc
 
+          - arch: arm
+            arm:  7
+            cc:   arm-linux-gnueabihf-gcc
+            static: true
+
     env:
       GOOS:   linux
       GOARCH: ${{ matrix.arch }}
       GOARM:  ${{ matrix.arm }}
       CC:     ${{ matrix.cc }}
+      LDFLAGS: ${{ matrix.static && '-extldflags "-static"' || '' }}
+      TAGS: ${{ matrix.static && 'osusergo,netgo,sqlite_omit_load_extension' || '' }}
+      SUFFIX: "${{ matrix.static && '-static' || ''}}"
+      VERSION: "${{ github.event_name == 'release' && github.event.release.name || github.sha }}"
+
 
     steps:
       - uses: actions/checkout@v2
+
       - uses: actions/setup-go@v2
         with:
-          go-version: '1.16'
+          go-version: '1.17'
-
-      - id: release
-        uses: bruceadams/get-release@v1.2.2
-        env:
-          GITHUB_TOKEN: ${{ github.token }}
 
       - name: Install cross-compilers
         run: |
@@ -50,32 +80,49 @@ jobs:
 
       - name: Build litestream
         run: |
-          rm -rf dist
+          rm -rf dist && mkdir -p dist
-          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
+          cat etc/nfpm.yml | LITESTREAM_VERSION=${{ env.VERSION }} envsubst > dist/nfpm.yml
-          CGO_ENABLED=1 go build -ldflags "-s -w -X 'main.Version=${{ steps.release.outputs.tag_name }}'" -o dist/litestream ./cmd/litestream
+
+          CGO_ENABLED=1 go build -ldflags "-s -w ${{ env.LDFLAGS }} -X 'main.Version=${{ env.VERSION }}'" -tags "${{ env.TAGS }}" -o dist/litestream ./cmd/litestream
 
           cd dist
-          tar -czvf litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.tar.gz litestream
+          tar -czvf litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.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-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.deb 
+
+      - name: Upload binary artifact
+        uses: actions/upload-artifact@v2
+        with:
+          name: litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.tar.gz
+          path: dist/litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.tar.gz
+          if-no-files-found: error
+
+      - name: Upload debian artifact
+        uses: actions/upload-artifact@v2
+        with:
+          name: litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.deb
+          path: dist/litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.deb
+          if-no-files-found: error
 
       - name: Upload release tarball
         uses: actions/upload-release-asset@v1.0.2
+        if: github.event_name == 'release'
         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 }}.tar.gz
+          asset_path: ./dist/litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.tar.gz
-          asset_name: litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.tar.gz
+          asset_name: litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.tar.gz
           asset_content_type: application/gzip
 
       - name: Upload debian package
         uses: actions/upload-release-asset@v1.0.2
+        if: github.event_name == 'release'
         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 }}.deb
+          asset_path: ./dist/litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.deb
-          asset_name: litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.deb
+          asset_name: litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.deb
           asset_content_type: application/octet-stream

.github/workflows/release.linux_static.yml

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

.github/workflows/test.yml

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

Dockerfile

@@ -1,11 +1,15 @@
-FROM golang:1.16 as builder
+FROM golang:1.17 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
 COPY --from=builder /usr/local/bin/litestream /usr/local/bin/litestream
 ENTRYPOINT ["/usr/local/bin/litestream"]

Makefile

@@ -1,5 +1,11 @@
+.PHONY: default
 default:
 
+.PHONY: testdata
+testdata:
+	make -C testdata
+	make -C cmd/litestream testdata
+
 docker:
 	docker build -t litestream .
 

README.md

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

abs/replica_client.go

@@ -190,8 +190,6 @@ func (c *ReplicaClient) WriteSnapshot(ctx context.Context, generation string, in
 	internal.OperationTotalCounterVec.WithLabelValues(ReplicaClientType, "PUT").Inc()
 	internal.OperationBytesCounterVec.WithLabelValues(ReplicaClientType, "PUT").Add(float64(rc.N()))
 
-	// log.Printf("%s(%s): snapshot: creating %s/%08x t=%s", r.db.Path(), r.Name(), generation, index, time.Since(startTime).Truncate(time.Millisecond))
-
 	return litestream.SnapshotInfo{
 		Generation: generation,
 		Index:      index,

cmd/litestream/Makefile

@@ -0,0 +1,6 @@
+.PHONY: default
+default:
+
+.PHONY: testdata
+testdata:
+	make -C testdata

cmd/litestream/databases.go

@@ -4,18 +4,34 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"os"
+	"io"
 	"strings"
 	"text/tabwriter"
 )
 
 // DatabasesCommand is a command for listing managed databases.
-type DatabasesCommand struct{}
+type DatabasesCommand struct {
+	stdin  io.Reader
+	stdout io.Writer
+	stderr io.Writer
+
+	configPath  string
+	noExpandEnv bool
+}
+
+// NewDatabasesCommand returns a new instance of DatabasesCommand.
+func NewDatabasesCommand(stdin io.Reader, stdout, stderr io.Writer) *DatabasesCommand {
+	return &DatabasesCommand{
+		stdin:  stdin,
+		stdout: stdout,
+		stderr: stderr,
+	}
+}
 
 // Run executes the command.
 func (c *DatabasesCommand) Run(ctx context.Context, args []string) (err error) {
 	fs := flag.NewFlagSet("litestream-databases", flag.ContinueOnError)
-	configPath, noExpandEnv := registerConfigFlag(fs)
+	registerConfigFlag(fs, &c.configPath, &c.noExpandEnv)
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -24,16 +40,16 @@ func (c *DatabasesCommand) Run(ctx context.Context, args []string) (err error) {
 	}
 
 	// Load configuration.
-	if *configPath == "" {
+	config, err := ReadConfigFile(c.configPath, !c.noExpandEnv)
-		*configPath = DefaultConfigPath()
-	}
-	config, err := ReadConfigFile(*configPath, !*noExpandEnv)
 	if err != nil {
 		return err
+	} else if len(config.DBs) == 0 {
+		fmt.Fprintln(c.stdout, "No databases found in config file.")
+		return nil
 	}
 
 	// List all databases.
-	w := tabwriter.NewWriter(os.Stdout, 0, 8, 2, ' ', 0)
+	w := tabwriter.NewWriter(c.stdout, 0, 8, 2, ' ', 0)
 	defer w.Flush()
 
 	fmt.Fprintln(w, "path\treplicas")
@@ -59,7 +75,7 @@ func (c *DatabasesCommand) Run(ctx context.Context, args []string) (err error) {
 
 // Usage prints the help screen to STDOUT.
 func (c *DatabasesCommand) Usage() {
-	fmt.Printf(`
+	fmt.Fprintf(c.stdout, `
 The databases command lists all databases in the configuration file.
 
 Usage:

cmd/litestream/databases_test.go

@@ -0,0 +1,66 @@
+package main_test
+
+import (
+	"context"
+	"flag"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/benbjohnson/litestream/internal/testingutil"
+)
+
+func TestDatabasesCommand(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "databases", "ok")
+		m, _, stdout, _ := newMain()
+		if err := m.Run(context.Background(), []string{"databases", "-config", filepath.Join(testDir, "litestream.yml")}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("NoDatabases", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "databases", "no-databases")
+		m, _, stdout, _ := newMain()
+		if err := m.Run(context.Background(), []string{"databases", "-config", filepath.Join(testDir, "litestream.yml")}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("ErrConfigNotFound", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "databases", "no-config")
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"databases", "-config", filepath.Join(testDir, "litestream.yml")})
+		if err == nil || !strings.Contains(err.Error(), `config file not found:`) {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrInvalidConfig", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "databases", "invalid-config")
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"databases", "-config", filepath.Join(testDir, "litestream.yml")})
+		if err == nil || !strings.Contains(err.Error(), `replica path cannot be a url`) {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrTooManyArguments", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"databases", "xyz"})
+		if err == nil || err.Error() != `too many arguments` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("Usage", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		if err := m.Run(context.Background(), []string{"databases", "-h"}); err != flag.ErrHelp {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+}

cmd/litestream/generations.go

@@ -4,117 +4,116 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"log"
+	"io"
 	"os"
 	"text/tabwriter"
 	"time"
 
 	"github.com/benbjohnson/litestream"
+	"github.com/benbjohnson/litestream/internal"
 )
 
 // GenerationsCommand represents a command to list all generations for a database.
-type GenerationsCommand struct{}
+type GenerationsCommand struct {
+	stdin  io.Reader
+	stdout io.Writer
+	stderr io.Writer
+
+	configPath  string
+	noExpandEnv bool
+
+	replicaName string
+}
+
+// NewGenerationsCommand returns a new instance of GenerationsCommand.
+func NewGenerationsCommand(stdin io.Reader, stdout, stderr io.Writer) *GenerationsCommand {
+	return &GenerationsCommand{
+		stdin:  stdin,
+		stdout: stdout,
+		stderr: stderr,
+	}
+}
 
 // Run executes the command.
-func (c *GenerationsCommand) Run(ctx context.Context, args []string) (err error) {
+func (c *GenerationsCommand) Run(ctx context.Context, args []string) (ret error) {
 	fs := flag.NewFlagSet("litestream-generations", flag.ContinueOnError)
-	configPath, noExpandEnv := registerConfigFlag(fs)
+	registerConfigFlag(fs, &c.configPath, &c.noExpandEnv)
-	replicaName := fs.String("replica", "", "replica name")
+	fs.StringVar(&c.replicaName, "replica", "", "replica name")
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
 		return err
-	} else if fs.NArg() == 0 || fs.Arg(0) == "" {
+	} else if fs.Arg(0) == "" {
 		return fmt.Errorf("database path or replica URL required")
 	} else if fs.NArg() > 1 {
 		return fmt.Errorf("too many arguments")
 	}
 
-	var db *litestream.DB
+	// Load configuration.
-	var r *litestream.Replica
+	config, err := ReadConfigFile(c.configPath, !c.noExpandEnv)
-	dbUpdatedAt := time.Now()
+	if err != nil {
-	if isURL(fs.Arg(0)) {
+		return err
-		if *configPath != "" {
-			return fmt.Errorf("cannot specify a replica URL and the -config flag")
-		}
-		if r, err = NewReplicaFromConfig(&ReplicaConfig{URL: fs.Arg(0)}, nil); err != nil {
-			return err
-		}
-	} else {
-		if *configPath == "" {
-			*configPath = DefaultConfigPath()
-		}
-
-		// Load configuration.
-		config, err := ReadConfigFile(*configPath, !*noExpandEnv)
-		if err != nil {
-			return err
-		}
-
-		// Lookup database from configuration file by path.
-		if path, err := expand(fs.Arg(0)); err != nil {
-			return err
-		} else if dbc := config.DBConfig(path); dbc == nil {
-			return fmt.Errorf("database not found in config: %s", path)
-		} else if db, err = NewDBFromConfig(dbc); err != nil {
-			return err
-		}
-
-		// Filter by replica, if specified.
-		if *replicaName != "" {
-			if r = db.Replica(*replicaName); r == nil {
-				return fmt.Errorf("replica %q not found for database %q", *replicaName, db.Path())
-			}
-		}
-
-		// Determine last time database or WAL was updated.
-		if dbUpdatedAt, err = db.UpdatedAt(); err != nil {
-			return err
-		}
 	}
 
-	var replicas []*litestream.Replica
+	replicas, db, err := loadReplicas(ctx, config, fs.Arg(0), c.replicaName)
-	if r != nil {
+	if err != nil {
-		replicas = []*litestream.Replica{r}
+		return err
-	} else {
+	}
-		replicas = db.Replicas
+
+	// Determine last time database or WAL was updated.
+	var dbUpdatedAt time.Time
+	if db != nil {
+		if dbUpdatedAt, err = db.UpdatedAt(); err != nil && !os.IsNotExist(err) {
+			return err
+		}
 	}
 
 	// List each generation.
-	w := tabwriter.NewWriter(os.Stdout, 0, 8, 2, ' ', 0)
+	w := tabwriter.NewWriter(c.stdout, 0, 8, 2, ' ', 0)
 	defer w.Flush()
 
 	fmt.Fprintln(w, "name\tgeneration\tlag\tstart\tend")
+
 	for _, r := range replicas {
-		generations, err := r.Client.Generations(ctx)
+		generations, err := r.Client().Generations(ctx)
 		if err != nil {
-			log.Printf("%s: cannot list generations: %s", r.Name(), err)
+			fmt.Fprintf(c.stderr, "%s: cannot list generations: %s", r.Name(), err)
+			ret = errExit // signal error return without printing message
 			continue
 		}
 
 		// Iterate over each generation for the replica.
 		for _, generation := range generations {
-			createdAt, updatedAt, err := r.GenerationTimeBounds(ctx, generation)
+			createdAt, updatedAt, err := litestream.GenerationTimeBounds(ctx, r.Client(), generation)
 			if err != nil {
-				log.Printf("%s: cannot determine generation time bounds: %s", r.Name(), err)
+				fmt.Fprintf(c.stderr, "%s: cannot determine generation time bounds: %s", r.Name(), err)
+				ret = errExit // signal error return without printing message
 				continue
 			}
 
+			// Calculate lag from database mod time to the replica mod time.
+			// This is ignored if the database mod time is unavailable such as
+			// when specifying the replica URL or if the database file is missing.
+			lag := "-"
+			if !dbUpdatedAt.IsZero() {
+				lag = internal.TruncateDuration(dbUpdatedAt.Sub(updatedAt)).String()
+			}
+
 			fmt.Fprintf(w, "%s\t%s\t%s\t%s\t%s\n",
 				r.Name(),
 				generation,
-				truncateDuration(dbUpdatedAt.Sub(updatedAt)).String(),
+				lag,
 				createdAt.Format(time.RFC3339),
 				updatedAt.Format(time.RFC3339),
 			)
 		}
 	}
 
-	return nil
+	return ret
 }
 
 // Usage prints the help message to STDOUT.
 func (c *GenerationsCommand) Usage() {
-	fmt.Printf(`
+	fmt.Fprintf(c.stdout, `
 The generations command lists all generations for a database or replica. It also
 lists stats about their lag behind the primary database and the time range they
 cover.
@@ -141,29 +140,3 @@ Arguments:
 		DefaultConfigPath(),
 	)
 }
-
-func truncateDuration(d time.Duration) time.Duration {
-	if d < 0 {
-		if d < -10*time.Second {
-			return d.Truncate(time.Second)
-		} else if d < -time.Second {
-			return d.Truncate(time.Second / 10)
-		} else if d < -time.Millisecond {
-			return d.Truncate(time.Millisecond)
-		} else if d < -time.Microsecond {
-			return d.Truncate(time.Microsecond)
-		}
-		return d
-	}
-
-	if d > 10*time.Second {
-		return d.Truncate(time.Second)
-	} else if d > time.Second {
-		return d.Truncate(time.Second / 10)
-	} else if d > time.Millisecond {
-		return d.Truncate(time.Millisecond)
-	} else if d > time.Microsecond {
-		return d.Truncate(time.Microsecond)
-	}
-	return d
-}

cmd/litestream/generations_test.go

@@ -0,0 +1,140 @@
+package main_test
+
+import (
+	"context"
+	"flag"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/benbjohnson/litestream/internal/testingutil"
+)
+
+func TestGenerationsCommand(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "generations", "ok")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, stdout, _ := newMain()
+		if err := m.Run(context.Background(), []string{"generations", "-config", filepath.Join(testDir, "litestream.yml"), filepath.Join(testDir, "db")}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("ReplicaName", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "generations", "replica-name")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, stdout, _ := newMain()
+		if err := m.Run(context.Background(), []string{"generations", "-config", filepath.Join(testDir, "litestream.yml"), "-replica", "replica1", filepath.Join(testDir, "db")}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("ReplicaURL", func(t *testing.T) {
+		testDir := filepath.Join(testingutil.Getwd(t), "testdata", "generations", "replica-url")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+		replicaURL := "file://" + filepath.ToSlash(testDir) + "/replica"
+
+		m, _, stdout, _ := newMain()
+		if err := m.Run(context.Background(), []string{"generations", replicaURL}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("NoDatabase", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "generations", "no-database")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, stdout, _ := newMain()
+		if err := m.Run(context.Background(), []string{"generations", "-config", filepath.Join(testDir, "litestream.yml"), filepath.Join(testDir, "db")}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("ErrDatabaseOrReplicaRequired", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"generations"})
+		if err == nil || err.Error() != `database path or replica URL required` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrTooManyArguments", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"generations", "abc", "123"})
+		if err == nil || err.Error() != `too many arguments` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrInvalidFlags", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"generations", "-no-such-flag"})
+		if err == nil || err.Error() != `flag provided but not defined: -no-such-flag` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrConfigFileNotFound", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"generations", "-config", "/no/such/file", "/var/lib/db"})
+		if err == nil || err.Error() != `config file not found: /no/such/file` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrInvalidConfig", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "generations", "invalid-config")
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"generations", "-config", filepath.Join(testDir, "litestream.yml"), "/var/lib/db"})
+		if err == nil || !strings.Contains(err.Error(), `replica path cannot be a url`) {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrDatabaseNotFound", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "generations", "database-not-found")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"generations", "-config", filepath.Join(testDir, "litestream.yml"), "/no/such/db"})
+		if err == nil || err.Error() != `database not found in config: /no/such/db` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrReplicaNotFound", func(t *testing.T) {
+		testDir := filepath.Join(testingutil.Getwd(t), "testdata", "generations", "replica-not-found")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"generations", "-config", filepath.Join(testDir, "litestream.yml"), "-replica", "no_such_replica", filepath.Join(testDir, "db")})
+		if err == nil || err.Error() != `replica "no_such_replica" not found for database "`+filepath.Join(testDir, "db")+`"` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrInvalidReplicaURL", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"generations", "xyz://xyz"})
+		if err == nil || !strings.Contains(err.Error(), `unknown replica type in config: "xyz"`) {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("Usage", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		if err := m.Run(context.Background(), []string{"generations", "-h"}); err != flag.ErrHelp {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+}

cmd/litestream/main.go

@@ -5,22 +5,25 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"log"
 	"net/url"
 	"os"
+	"os/signal"
 	"os/user"
 	"path"
 	"path/filepath"
 	"regexp"
 	"strconv"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/abs"
-	"github.com/benbjohnson/litestream/file"
+	"github.com/benbjohnson/litestream/gs"
-	"github.com/benbjohnson/litestream/gcs"
+	"github.com/benbjohnson/litestream/http"
 	"github.com/benbjohnson/litestream/s3"
 	"github.com/benbjohnson/litestream/sftp"
 	_ "github.com/mattn/go-sqlite3"
@@ -32,14 +35,15 @@ var (
 	Version = "(development build)"
 )
 
-// errStop is a terminal error for indicating program should quit.
+// errExit is a terminal error for indicating program should quit.
-var errStop = errors.New("stop")
+var errExit = errors.New("exit")
 
 func main() {
 	log.SetFlags(0)
+	log.SetOutput(os.Stdout)
 
-	m := NewMain()
+	m := NewMain(os.Stdin, os.Stdout, os.Stderr)
-	if err := m.Run(context.Background(), os.Args[1:]); err == flag.ErrHelp || err == errStop {
+	if err := m.Run(context.Background(), os.Args[1:]); err == flag.ErrHelp || err == errExit {
 		os.Exit(1)
 	} else if err != nil {
 		log.Println(err)
@@ -48,22 +52,23 @@ func main() {
 }
 
 // Main represents the main program execution.
-type Main struct{}
+type Main struct {
+	stdin  io.Reader
+	stdout io.Writer
+	stderr io.Writer
+}
 
 // NewMain returns a new instance of Main.
-func NewMain() *Main {
+func NewMain(stdin io.Reader, stdout, stderr io.Writer) *Main {
-	return &Main{}
+	return &Main{
+		stdin:  stdin,
+		stdout: stdout,
+		stderr: stderr,
+	}
 }
 
 // Run executes the program.
 func (m *Main) Run(ctx context.Context, args []string) (err error) {
-	// Execute replication command if running as a Windows service.
-	if isService, err := isWindowsService(); err != nil {
-		return err
-	} else if isService {
-		return runWindowsService(ctx)
-	}
-
 	// Copy "LITESTEAM" environment credentials.
 	applyLitestreamEnv()
 
@@ -75,18 +80,20 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 
 	switch cmd {
 	case "databases":
-		return (&DatabasesCommand{}).Run(ctx, args)
+		return NewDatabasesCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
 	case "generations":
-		return (&GenerationsCommand{}).Run(ctx, args)
+		return NewGenerationsCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
 	case "replicate":
-		c := NewReplicateCommand()
+		c := NewReplicateCommand(m.stdin, m.stdout, m.stderr)
 		if err := c.ParseFlags(ctx, args); err != nil {
 			return err
 		}
 
 		// Setup signal handler.
 		ctx, cancel := context.WithCancel(ctx)
-		signalCh := signalChan()
+		defer cancel()
+		signalCh := make(chan os.Signal, 1)
+		signal.Notify(signalCh, syscall.SIGINT, syscall.SIGTERM)
 
 		if err := c.Run(ctx); err != nil {
 			return err
@@ -94,20 +101,22 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 
 		// Wait for signal to stop program.
 		select {
+		case <-ctx.Done():
+			fmt.Fprintln(m.stdout, "context done, litestream shutting down")
 		case err = <-c.execCh:
 			cancel()
-			fmt.Println("subprocess exited, litestream shutting down")
+			fmt.Fprintln(m.stdout, "subprocess exited, litestream shutting down")
 		case sig := <-signalCh:
 			cancel()
-			fmt.Println("signal received, litestream shutting down")
+			fmt.Fprintln(m.stdout, "signal received, litestream shutting down")
 
 			if c.cmd != nil {
-				fmt.Println("sending signal to exec process")
+				fmt.Fprintln(m.stdout, "sending signal to exec process")
 				if err := c.cmd.Process.Signal(sig); err != nil {
 					return fmt.Errorf("cannot signal exec process: %w", err)
 				}
 
-				fmt.Println("waiting for exec process to close")
+				fmt.Fprintln(m.stdout, "waiting for exec process to close")
 				if err := <-c.execCh; err != nil && !strings.HasPrefix(err.Error(), "signal:") {
 					return fmt.Errorf("cannot wait for exec process: %w", err)
 				}
@@ -118,17 +127,17 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 		if e := c.Close(); e != nil && err == nil {
 			err = e
 		}
-		fmt.Println("litestream shut down")
+		fmt.Fprintln(m.stdout, "litestream shut down")
 		return err
 
 	case "restore":
-		return (&RestoreCommand{}).Run(ctx, args)
+		return NewRestoreCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
 	case "snapshots":
-		return (&SnapshotsCommand{}).Run(ctx, args)
+		return NewSnapshotsCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
 	case "version":
-		return (&VersionCommand{}).Run(ctx, args)
+		return NewVersionCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
 	case "wal":
-		return (&WALCommand{}).Run(ctx, args)
+		return NewWALCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
 	default:
 		if cmd == "" || cmd == "help" || strings.HasPrefix(cmd, "-") {
 			m.Usage()
@@ -140,7 +149,7 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 
 // Usage prints the help screen to STDOUT.
 func (m *Main) Usage() {
-	fmt.Println(`
+	fmt.Fprintln(m.stdout, `
 litestream is a tool for replicating SQLite databases.
 
 Usage:
@@ -207,7 +216,34 @@ func (c *Config) DBConfig(path string) *DBConfig {
 
 // ReadConfigFile unmarshals config from filename. Expands path if needed.
 // If expandEnv is true then environment variables are expanded in the config.
-func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
+// If filename is blank then the default config path is used.
+func ReadConfigFile(filename string, expandEnv bool) (config Config, err error) {
+	var filenames []string
+	if filename != "" {
+		filenames = append(filenames, filename)
+	}
+	filenames = append(filenames, "./litestream.yml")
+	filenames = append(filenames, DefaultConfigPath())
+
+	for _, name := range filenames {
+		isDefaultPath := name != filename
+
+		if config, err = readConfigFile(name, expandEnv); os.IsNotExist(err) {
+			if isDefaultPath {
+				continue
+			}
+			return config, fmt.Errorf("config file not found: %s", filename)
+		} else if err != nil {
+			return config, err
+		}
+		break
+	}
+	return config, nil
+}
+
+// readConfigFile unmarshals config from filename. Expands path if needed.
+// If expandEnv is true then environment variables are expanded in the config.
+func readConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 	config := DefaultConfig()
 
 	// Expand filename, if necessary.
@@ -217,10 +253,9 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 	}
 
 	// Read configuration.
+	// Do not return an error if using default path and file is missing.
 	buf, err := ioutil.ReadFile(filename)
-	if os.IsNotExist(err) {
+	if err != nil {
-		return config, fmt.Errorf("config file not found: %s", filename)
-	} else if err != nil {
 		return config, err
 	}
 
@@ -248,11 +283,13 @@ func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 
 // DBConfig represents the configuration for a single database.
 type DBConfig struct {
-	Path               string         `yaml:"path"`
+	Path                 string         `yaml:"path"`
-	MonitorInterval    *time.Duration `yaml:"monitor-interval"`
+	Upstream             UpstreamConfig `yaml:"upstream"`
-	CheckpointInterval *time.Duration `yaml:"checkpoint-interval"`
+	MonitorDelayInterval *time.Duration `yaml:"monitor-delay-interval"`
-	MinCheckpointPageN *int           `yaml:"min-checkpoint-page-count"`
+	CheckpointInterval   *time.Duration `yaml:"checkpoint-interval"`
-	MaxCheckpointPageN *int           `yaml:"max-checkpoint-page-count"`
+	MinCheckpointPageN   *int           `yaml:"min-checkpoint-page-count"`
+	MaxCheckpointPageN   *int           `yaml:"max-checkpoint-page-count"`
+	ShadowRetentionN     *int           `yaml:"shadow-retention-count"`
 
 	Replicas []*ReplicaConfig `yaml:"replicas"`
 }
@@ -263,13 +300,27 @@ func NewDBFromConfig(dbc *DBConfig) (*litestream.DB, error) {
 	if err != nil {
 		return nil, err
 	}
+	return NewDBFromConfigWithPath(dbc, path)
+}
 
+// NewDBFromConfigWithPath instantiates a DB based on a configuration and using a given path.
+func NewDBFromConfigWithPath(dbc *DBConfig, path string) (*litestream.DB, error) {
 	// Initialize database with given path.
 	db := litestream.NewDB(path)
 
+	// Attach upstream HTTP client if specified.
+	if upstreamURL := dbc.Upstream.URL; upstreamURL != "" {
+		// Use local database path if upstream path is not specified.
+		upstreamPath := dbc.Upstream.Path
+		if upstreamPath == "" {
+			upstreamPath = db.Path()
+		}
+		db.StreamClient = http.NewClient(upstreamURL, upstreamPath)
+	}
+
 	// Override default database settings if specified in configuration.
-	if dbc.MonitorInterval != nil {
+	if dbc.MonitorDelayInterval != nil {
-		db.MonitorInterval = *dbc.MonitorInterval
+		db.MonitorDelayInterval = *dbc.MonitorDelayInterval
 	}
 	if dbc.CheckpointInterval != nil {
 		db.CheckpointInterval = *dbc.CheckpointInterval
@@ -280,6 +331,9 @@ func NewDBFromConfig(dbc *DBConfig) (*litestream.DB, error) {
 	if dbc.MaxCheckpointPageN != nil {
 		db.MaxCheckpointPageN = *dbc.MaxCheckpointPageN
 	}
+	if dbc.ShadowRetentionN != nil {
+		db.ShadowRetentionN = *dbc.ShadowRetentionN
+	}
 
 	// Instantiate and attach replicas.
 	for _, rc := range dbc.Replicas {
@@ -293,6 +347,11 @@ func NewDBFromConfig(dbc *DBConfig) (*litestream.DB, error) {
 	return db, nil
 }
 
+type UpstreamConfig struct {
+	URL  string `yaml:"url"`
+	Path string `yaml:"path"`
+}
+
 // ReplicaConfig represents the configuration for a single replica in a database.
 type ReplicaConfig struct {
 	Type                   string         `yaml:"type"` // "file", "s3"
@@ -332,8 +391,35 @@ func NewReplicaFromConfig(c *ReplicaConfig, db *litestream.DB) (_ *litestream.Re
 		return nil, fmt.Errorf("replica path cannot be a url, please use the 'url' field instead: %s", c.Path)
 	}
 
+	// Build and set client on replica.
+	var client litestream.ReplicaClient
+	switch typ := c.ReplicaType(); typ {
+	case "file":
+		if client, err = newFileReplicaClientFromConfig(c); err != nil {
+			return nil, err
+		}
+	case "s3":
+		if client, err = newS3ReplicaClientFromConfig(c); err != nil {
+			return nil, err
+		}
+	case "gs":
+		if client, err = newGSReplicaClientFromConfig(c); err != nil {
+			return nil, err
+		}
+	case "abs":
+		if client, err = newABSReplicaClientFromConfig(c); err != nil {
+			return nil, err
+		}
+	case "sftp":
+		if client, err = newSFTPReplicaClientFromConfig(c); err != nil {
+			return nil, err
+		}
+	default:
+		return nil, fmt.Errorf("unknown replica type in config: %q", typ)
+	}
+
 	// Build replica.
-	r := litestream.NewReplica(db, c.Name)
+	r := litestream.NewReplica(db, c.Name, client)
 	if v := c.Retention; v != nil {
 		r.Retention = *v
 	}
@@ -350,37 +436,11 @@ func NewReplicaFromConfig(c *ReplicaConfig, db *litestream.DB) (_ *litestream.Re
 		r.ValidationInterval = *v
 	}
 
-	// Build and set client on replica.
-	switch c.ReplicaType() {
-	case "file":
-		if r.Client, err = newFileReplicaClientFromConfig(c, r); err != nil {
-			return nil, err
-		}
-	case "s3":
-		if r.Client, err = newS3ReplicaClientFromConfig(c, r); err != nil {
-			return nil, err
-		}
-	case "gcs":
-		if r.Client, err = newGCSReplicaClientFromConfig(c, r); err != nil {
-			return nil, err
-		}
-	case "abs":
-		if r.Client, err = newABSReplicaClientFromConfig(c, r); err != nil {
-			return nil, err
-		}
-	case "sftp":
-		if r.Client, err = newSFTPReplicaClientFromConfig(c, r); err != nil {
-			return nil, err
-		}
-	default:
-		return nil, fmt.Errorf("unknown replica type in config: %q", c.Type)
-	}
-
 	return r, nil
 }
 
-// newFileReplicaClientFromConfig returns a new instance of file.ReplicaClient built from config.
+// newFileReplicaClientFromConfig returns a new instance of FileReplicaClient built from config.
-func newFileReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *file.ReplicaClient, err error) {
+func newFileReplicaClientFromConfig(c *ReplicaConfig) (_ *litestream.FileReplicaClient, err error) {
 	// Ensure URL & path are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for file replica")
@@ -405,13 +465,11 @@ func newFileReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_
 	}
 
 	// Instantiate replica and apply time fields, if set.
-	client := file.NewReplicaClient(path)
+	return litestream.NewFileReplicaClient(path), nil
-	client.Replica = r
-	return client, nil
 }
 
 // newS3ReplicaClientFromConfig returns a new instance of s3.ReplicaClient built from config.
-func newS3ReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *s3.ReplicaClient, err error) {
+func newS3ReplicaClientFromConfig(c *ReplicaConfig) (_ *s3.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for s3 replica")
@@ -473,13 +531,13 @@ func newS3ReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *s
 	return client, nil
 }
 
-// newGCSReplicaClientFromConfig returns a new instance of gcs.ReplicaClient built from config.
+// newGSReplicaClientFromConfig returns a new instance of gs.ReplicaClient built from config.
-func newGCSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *gcs.ReplicaClient, err error) {
+func newGSReplicaClientFromConfig(c *ReplicaConfig) (_ *gs.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
-		return nil, fmt.Errorf("cannot specify url & path for gcs replica")
+		return nil, fmt.Errorf("cannot specify url & path for gs replica")
 	} else if c.URL != "" && c.Bucket != "" {
-		return nil, fmt.Errorf("cannot specify url & bucket for gcs replica")
+		return nil, fmt.Errorf("cannot specify url & bucket for gs replica")
 	}
 
 	bucket, path := c.Bucket, c.Path
@@ -502,18 +560,18 @@ func newGCSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *
 
 	// Ensure required settings are set.
 	if bucket == "" {
-		return nil, fmt.Errorf("bucket required for gcs replica")
+		return nil, fmt.Errorf("bucket required for gs replica")
 	}
 
 	// Build replica.
-	client := gcs.NewReplicaClient()
+	client := gs.NewReplicaClient()
 	client.Bucket = bucket
 	client.Path = path
 	return client, nil
 }
 
 // newABSReplicaClientFromConfig returns a new instance of abs.ReplicaClient built from config.
-func newABSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *abs.ReplicaClient, err error) {
+func newABSReplicaClientFromConfig(c *ReplicaConfig) (_ *abs.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for abs replica")
@@ -556,7 +614,7 @@ func newABSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *
 }
 
 // newSFTPReplicaClientFromConfig returns a new instance of sftp.ReplicaClient built from config.
-func newSFTPReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *sftp.ReplicaClient, err error) {
+func newSFTPReplicaClientFromConfig(c *ReplicaConfig) (_ *sftp.ReplicaClient, err error) {
 	// Ensure URL & constituent parts are not both specified.
 	if c.URL != "" && c.Path != "" {
 		return nil, fmt.Errorf("cannot specify url & path for sftp replica")
@@ -662,12 +720,12 @@ func DefaultConfigPath() string {
 	if v := os.Getenv("LITESTREAM_CONFIG"); v != "" {
 		return v
 	}
-	return defaultConfigPath
+	return "/etc/litestream.yml"
 }
 
-func registerConfigFlag(fs *flag.FlagSet) (configPath *string, noExpandEnv *bool) {
+func registerConfigFlag(fs *flag.FlagSet, configPath *string, noExpandEnv *bool) {
-	return fs.String("config", "", "config path"),
+	fs.StringVar(configPath, "config", "", "config path")
-		fs.Bool("no-expand-env", false, "do not expand env vars in config")
+	fs.BoolVar(noExpandEnv, "no-expand-env", false, "do not expand env vars in config")
 }
 
 // expand returns an absolute path for s.
@@ -701,7 +759,7 @@ var _ flag.Value = (*indexVar)(nil)
 
 // String returns an 8-character hexadecimal value.
 func (v *indexVar) String() string {
-	return fmt.Sprintf("%08x", int(*v))
+	return litestream.FormatIndex(int(*v))
 }
 
 // Set parses s into an integer from a hexadecimal value.
@@ -713,3 +771,45 @@ func (v *indexVar) Set(s string) error {
 	*v = indexVar(i)
 	return nil
 }
+
+// loadReplicas returns a list of replicas to use based on CLI flags. Filters
+// by replicaName, if not blank. The DB is returned if pathOrURL is not a replica URL.
+func loadReplicas(ctx context.Context, config Config, pathOrURL, replicaName string) ([]*litestream.Replica, *litestream.DB, error) {
+	// Build a replica based on URL, if specified.
+	if isURL(pathOrURL) {
+		r, err := NewReplicaFromConfig(&ReplicaConfig{
+			URL:             pathOrURL,
+			AccessKeyID:     config.AccessKeyID,
+			SecretAccessKey: config.SecretAccessKey,
+		}, nil)
+		if err != nil {
+			return nil, nil, err
+		}
+		return []*litestream.Replica{r}, nil, nil
+	}
+
+	// Otherwise use replicas from the database configuration file.
+	path, err := expand(pathOrURL)
+	if err != nil {
+		return nil, nil, err
+	}
+	dbc := config.DBConfig(path)
+	if dbc == nil {
+		return nil, nil, fmt.Errorf("database not found in config: %s", path)
+	}
+	db, err := NewDBFromConfig(dbc)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Filter by replica, if specified.
+	if replicaName != "" {
+		r := db.Replica(replicaName)
+		if r == nil {
+			return nil, nil, fmt.Errorf("replica %q not found for database %q", replicaName, db.Path())
+		}
+		return []*litestream.Replica{r}, db, nil
+	}
+
+	return db.Replicas, db, nil
+}

cmd/litestream/main_notwindows.go

@@ -1,26 +0,0 @@
-// +build !windows
-
-package main
-
-import (
-	"context"
-	"os"
-	"os/signal"
-	"syscall"
-)
-
-const defaultConfigPath = "/etc/litestream.yml"
-
-func isWindowsService() (bool, error) {
-	return false, nil
-}
-
-func runWindowsService(ctx context.Context) error {
-	panic("cannot run windows service as unix process")
-}
-
-func signalChan() <-chan os.Signal {
-	ch := make(chan os.Signal, 2)
-	signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
-	return ch
-}

cmd/litestream/main_test.go

@@ -1,17 +1,24 @@
 package main_test
 
 import (
+	"bytes"
+	"io"
 	"io/ioutil"
+	"log"
 	"os"
 	"path/filepath"
 	"testing"
 
+	"github.com/benbjohnson/litestream"
 	main "github.com/benbjohnson/litestream/cmd/litestream"
-	"github.com/benbjohnson/litestream/file"
+	"github.com/benbjohnson/litestream/gs"
-	"github.com/benbjohnson/litestream/gcs"
 	"github.com/benbjohnson/litestream/s3"
 )
 
+func init() {
+	litestream.LogFlags = log.Lmsgprefix | log.Ldate | log.Ltime | log.Lmicroseconds | log.LUTC | log.Lshortfile
+}
+
 func TestReadConfigFile(t *testing.T) {
 	// Ensure global AWS settings are propagated down to replica configurations.
 	t.Run("PropagateGlobalSettings", func(t *testing.T) {
@@ -97,7 +104,7 @@ func TestNewFileReplicaFromConfig(t *testing.T) {
 	r, err := main.NewReplicaFromConfig(&main.ReplicaConfig{Path: "/foo"}, nil)
 	if err != nil {
 		t.Fatal(err)
-	} else if client, ok := r.Client.(*file.ReplicaClient); !ok {
+	} else if client, ok := r.Client().(*litestream.FileReplicaClient); !ok {
 		t.Fatal("unexpected replica type")
 	} else if got, want := client.Path(), "/foo"; got != want {
 		t.Fatalf("Path=%s, want %s", got, want)
@@ -109,7 +116,7 @@ func TestNewS3ReplicaFromConfig(t *testing.T) {
 		r, err := main.NewReplicaFromConfig(&main.ReplicaConfig{URL: "s3://foo/bar"}, nil)
 		if err != nil {
 			t.Fatal(err)
-		} else if client, ok := r.Client.(*s3.ReplicaClient); !ok {
+		} else if client, ok := r.Client().(*s3.ReplicaClient); !ok {
 			t.Fatal("unexpected replica type")
 		} else if got, want := client.Bucket, "foo"; got != want {
 			t.Fatalf("Bucket=%s, want %s", got, want)
@@ -128,7 +135,7 @@ func TestNewS3ReplicaFromConfig(t *testing.T) {
 		r, err := main.NewReplicaFromConfig(&main.ReplicaConfig{URL: "s3://foo.localhost:9000/bar"}, nil)
 		if err != nil {
 			t.Fatal(err)
-		} else if client, ok := r.Client.(*s3.ReplicaClient); !ok {
+		} else if client, ok := r.Client().(*s3.ReplicaClient); !ok {
 			t.Fatal("unexpected replica type")
 		} else if got, want := client.Bucket, "foo"; got != want {
 			t.Fatalf("Bucket=%s, want %s", got, want)
@@ -147,7 +154,7 @@ func TestNewS3ReplicaFromConfig(t *testing.T) {
 		r, err := main.NewReplicaFromConfig(&main.ReplicaConfig{URL: "s3://foo.s3.us-west-000.backblazeb2.com/bar"}, nil)
 		if err != nil {
 			t.Fatal(err)
-		} else if client, ok := r.Client.(*s3.ReplicaClient); !ok {
+		} else if client, ok := r.Client().(*s3.ReplicaClient); !ok {
 			t.Fatal("unexpected replica type")
 		} else if got, want := client.Bucket, "foo"; got != want {
 			t.Fatalf("Bucket=%s, want %s", got, want)
@@ -163,11 +170,11 @@ func TestNewS3ReplicaFromConfig(t *testing.T) {
 	})
 }
 
-func TestNewGCSReplicaFromConfig(t *testing.T) {
+func TestNewGSReplicaFromConfig(t *testing.T) {
-	r, err := main.NewReplicaFromConfig(&main.ReplicaConfig{URL: "gcs://foo/bar"}, nil)
+	r, err := main.NewReplicaFromConfig(&main.ReplicaConfig{URL: "gs://foo/bar"}, nil)
 	if err != nil {
 		t.Fatal(err)
-	} else if client, ok := r.Client.(*gcs.ReplicaClient); !ok {
+	} else if client, ok := r.Client().(*gs.ReplicaClient); !ok {
 		t.Fatal("unexpected replica type")
 	} else if got, want := client.Bucket, "foo"; got != want {
 		t.Fatalf("Bucket=%s, want %s", got, want)
@@ -175,3 +182,17 @@ func TestNewGCSReplicaFromConfig(t *testing.T) {
 		t.Fatalf("Path=%s, want %s", got, want)
 	}
 }
+
+// newMain returns a new instance of Main and associated buffers.
+func newMain() (m *main.Main, stdin, stdout, stderr *bytes.Buffer) {
+	stdin, stdout, stderr = &bytes.Buffer{}, &bytes.Buffer{}, &bytes.Buffer{}
+
+	// Split stdout/stderr to terminal if verbose flag set.
+	out, err := io.Writer(stdout), io.Writer(stderr)
+	if testing.Verbose() {
+		out = io.MultiWriter(out, os.Stdout)
+		err = io.MultiWriter(err, os.Stderr)
+	}
+
+	return main.NewMain(stdin, out, err), stdin, stdout, stderr
+}

cmd/litestream/main_windows.go

@@ -1,112 +0,0 @@
-// +build windows
-
-package main
-
-import (
-	"context"
-	"io"
-	"log"
-	"os"
-	"os/signal"
-
-	"golang.org/x/sys/windows"
-	"golang.org/x/sys/windows/svc"
-	"golang.org/x/sys/windows/svc/eventlog"
-)
-
-const defaultConfigPath = `C:\Litestream\litestream.yml`
-
-// serviceName is the Windows Service name.
-const serviceName = "Litestream"
-
-// isWindowsService returns true if currently executing within a Windows service.
-func isWindowsService() (bool, error) {
-	return svc.IsWindowsService()
-}
-
-func runWindowsService(ctx context.Context) error {
-	// Attempt to install new log service. This will fail if already installed.
-	// We don't log the error because we don't have anywhere to log until we open the log.
-	_ = eventlog.InstallAsEventCreate(serviceName, eventlog.Error|eventlog.Warning|eventlog.Info)
-
-	elog, err := eventlog.Open(serviceName)
-	if err != nil {
-		return err
-	}
-	defer elog.Close()
-
-	// Set eventlog as log writer while running.
-	log.SetOutput((*eventlogWriter)(elog))
-	defer log.SetOutput(os.Stderr)
-
-	log.Print("Litestream service starting")
-
-	if err := svc.Run(serviceName, &windowsService{ctx: ctx}); err != nil {
-		return errStop
-	}
-
-	log.Print("Litestream service stopped")
-	return nil
-}
-
-// windowsService is an interface adapter for svc.Handler.
-type windowsService struct {
-	ctx context.Context
-}
-
-func (s *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, statusCh chan<- svc.Status) (svcSpecificEC bool, exitCode uint32) {
-	var err error
-
-	// Notify Windows that the service is starting up.
-	statusCh <- svc.Status{State: svc.StartPending}
-
-	// Instantiate replication command and load configuration.
-	c := NewReplicateCommand()
-	if c.Config, err = ReadConfigFile(DefaultConfigPath(), true); err != nil {
-		log.Printf("cannot load configuration: %s", err)
-		return true, 1
-	}
-
-	// Execute replication command.
-	if err := c.Run(s.ctx); err != nil {
-		log.Printf("cannot replicate: %s", err)
-		statusCh <- svc.Status{State: svc.StopPending}
-		return true, 2
-	}
-
-	// Notify Windows that the service is now running.
-	statusCh <- svc.Status{State: svc.Running, Accepts: svc.AcceptStop}
-
-	for {
-		select {
-		case req := <-r:
-			switch req.Cmd {
-			case svc.Stop:
-				c.Close()
-				statusCh <- svc.Status{State: svc.StopPending}
-				return false, windows.NO_ERROR
-			case svc.Interrogate:
-				statusCh <- req.CurrentStatus
-			default:
-				log.Printf("Litestream service received unexpected change request cmd: %d", req.Cmd)
-			}
-		}
-	}
-}
-
-// Ensure implementation implements io.Writer interface.
-var _ io.Writer = (*eventlogWriter)(nil)
-
-// eventlogWriter is an adapter for using eventlog.Log as an io.Writer.
-type eventlogWriter eventlog.Log
-
-func (w *eventlogWriter) Write(p []byte) (n int, err error) {
-	elog := (*eventlog.Log)(w)
-	return 0, elog.Info(1, string(p))
-}
-
-func signalChan() <-chan os.Signal {
-	ch := make(chan os.Signal, 1)
-	signal.Notify(ch, os.Interrupt)
-	return ch
-}

cmd/litestream/replicate.go

@@ -4,36 +4,45 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"io"
 	"log"
-	"net"
-	"net/http"
-	_ "net/http/pprof"
 	"os"
 	"os/exec"
 
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/abs"
-	"github.com/benbjohnson/litestream/file"
+	"github.com/benbjohnson/litestream/gs"
-	"github.com/benbjohnson/litestream/gcs"
+	"github.com/benbjohnson/litestream/http"
 	"github.com/benbjohnson/litestream/s3"
 	"github.com/benbjohnson/litestream/sftp"
 	"github.com/mattn/go-shellwords"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 )
 
 // ReplicateCommand represents a command that continuously replicates SQLite databases.
 type ReplicateCommand struct {
+	stdin  io.Reader
+	stdout io.Writer
+	stderr io.Writer
+
+	configPath  string
+	noExpandEnv bool
+
 	cmd    *exec.Cmd  // subcommand
 	execCh chan error // subcommand error channel
 
 	Config Config
 
-	// List of managed databases specified in the config.
+	server     *litestream.Server
-	DBs []*litestream.DB
+	httpServer *http.Server
 }
 
-func NewReplicateCommand() *ReplicateCommand {
+// NewReplicateCommand returns a new instance of ReplicateCommand.
+func NewReplicateCommand(stdin io.Reader, stdout, stderr io.Writer) *ReplicateCommand {
 	return &ReplicateCommand{
+		stdin:  stdin,
+		stdout: stdout,
+		stderr: stderr,
+
 		execCh: make(chan error),
 	}
 }
@@ -42,8 +51,8 @@ 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")
+	addr := fs.String("addr", "", "HTTP bind address (host:port)")
-	configPath, noExpandEnv := registerConfigFlag(fs)
+	registerConfigFlag(fs, &c.configPath, &c.noExpandEnv)
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -53,7 +62,7 @@ func (c *ReplicateCommand) ParseFlags(ctx context.Context, args []string) (err e
 	if fs.NArg() == 1 {
 		return fmt.Errorf("must specify at least one replica URL for %s", fs.Arg(0))
 	} else if fs.NArg() > 1 {
-		if *configPath != "" {
+		if c.configPath != "" {
 			return fmt.Errorf("cannot specify a replica URL and the -config flag")
 		}
 
@@ -67,29 +76,22 @@ func (c *ReplicateCommand) ParseFlags(ctx context.Context, args []string) (err e
 		}
 		c.Config.DBs = []*DBConfig{dbConfig}
 	} else {
-		if *configPath == "" {
+		if c.configPath == "" {
-			*configPath = DefaultConfigPath()
+			c.configPath = DefaultConfigPath()
 		}
-		if c.Config, err = ReadConfigFile(*configPath, !*noExpandEnv); err != nil {
+		if c.Config, err = ReadConfigFile(c.configPath, !c.noExpandEnv); err != nil {
 			return err
 		}
 	}
 
-	// Override config exec command, if specified.
+	// Override config with flags, if specified.
+	if *addr != "" {
+		c.Config.Addr = *addr
+	}
 	if *execFlag != "" {
 		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
 }
 
@@ -103,29 +105,35 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 		log.Println("no databases specified in configuration")
 	}
 
+	c.server = litestream.NewServer()
+	if err := c.server.Open(); err != nil {
+		return fmt.Errorf("open server: %w", err)
+	}
+
+	// Add databases to the server.
 	for _, dbConfig := range c.Config.DBs {
-		db, err := NewDBFromConfig(dbConfig)
+		path, err := expand(dbConfig.Path)
 		if err != nil {
 			return err
 		}
 
-		// Open database & attach to program.
+		if err := c.server.Watch(path, func(path string) (*litestream.DB, error) {
-		if err := db.Open(); err != nil {
+			return NewDBFromConfigWithPath(dbConfig, path)
+		}); err != nil {
 			return err
 		}
-		c.DBs = append(c.DBs, db)
 	}
 
 	// Notify user that initialization is done.
-	for _, db := range c.DBs {
+	for _, db := range c.server.DBs() {
 		log.Printf("initialized db: %s", db.Path())
 		for _, r := range db.Replicas {
-			switch client := r.Client.(type) {
+			switch client := r.Client().(type) {
-			case *file.ReplicaClient:
+			case *litestream.FileReplicaClient:
 				log.Printf("replicating to: name=%q type=%q path=%q", r.Name(), client.Type(), client.Path())
 			case *s3.ReplicaClient:
 				log.Printf("replicating to: name=%q type=%q bucket=%q path=%q region=%q endpoint=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, client.Region, client.Endpoint, r.SyncInterval)
-			case *gcs.ReplicaClient:
+			case *gs.ReplicaClient:
 				log.Printf("replicating to: name=%q type=%q bucket=%q path=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, r.SyncInterval)
 			case *abs.ReplicaClient:
 				log.Printf("replicating to: name=%q type=%q bucket=%q path=%q endpoint=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, client.Endpoint, r.SyncInterval)
@@ -137,22 +145,13 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 		}
 	}
 
-	// Serve metrics over HTTP if enabled.
+	// Serve HTTP if enabled.
 	if c.Config.Addr != "" {
-		hostport := c.Config.Addr
+		c.httpServer = http.NewServer(c.server, c.Config.Addr)
-		if host, port, _ := net.SplitHostPort(c.Config.Addr); port == "" {
+		if err := c.httpServer.Open(); err != nil {
-			return fmt.Errorf("must specify port for bind address: %q", c.Config.Addr)
+			return fmt.Errorf("cannot start http server: %w", err)
-		} else if host == "" {
-			hostport = net.JoinHostPort("localhost", port)
 		}
-
+		log.Printf("http server running at %s", c.httpServer.URL())
-		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 {
-				log.Printf("cannot start metrics server: %s", err)
-			}
-		}()
 	}
 
 	// Parse exec commands args & start subprocess.
@@ -162,8 +161,14 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 			return fmt.Errorf("cannot parse exec command: %w", err)
 		}
 
+		// Pass first database path to child process.
+		env := os.Environ()
+		if dbs := c.server.DBs(); len(dbs) > 0 {
+			env = append(env, fmt.Sprintf("LITESTREAM_DB_PATH=%s", dbs[0].Path()))
+		}
+
 		c.cmd = exec.CommandContext(ctx, execArgs[0], execArgs[1:]...)
-		c.cmd.Env = os.Environ()
+		c.cmd.Env = env
 		c.cmd.Stdout = os.Stdout
 		c.cmd.Stderr = os.Stderr
 		if err := c.cmd.Start(); err != nil {
@@ -172,17 +177,21 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 		go func() { c.execCh <- c.cmd.Wait() }()
 	}
 
+	log.Printf("litestream initialization complete")
+
 	return nil
 }
 
-// Close closes all open databases.
+// Close closes the HTTP server & all open databases.
 func (c *ReplicateCommand) Close() (err error) {
-	for _, db := range c.DBs {
+	if c.httpServer != nil {
-		if e := db.SoftClose(); e != nil {
+		if e := c.httpServer.Close(); e != nil && err == nil {
-			log.Printf("error closing db: path=%s err=%s", db.Path(), e)
+			err = e
-			if err == nil {
+		}
-				err = e
+	}
-			}
+	if c.server != nil {
+		if e := c.server.Close(); e != nil && err == nil {
+			err = e
 		}
 	}
 	return err
@@ -190,7 +199,7 @@ func (c *ReplicateCommand) Close() (err error) {
 
 // Usage prints the help screen to STDOUT.
 func (c *ReplicateCommand) Usage() {
-	fmt.Printf(`
+	fmt.Fprintf(c.stdout, `
 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
@@ -212,11 +221,12 @@ Arguments:
 	    Executes a subcommand. Litestream will exit when the child
 	    process exits. Useful for simple process management.
 
+	-addr BIND_ADDR
+	    Starts an HTTP server that reports prometheus metrics and provides
+	    an endpoint for live read replication. (e.g. ":9090")
+
 	-no-expand-env
 	    Disables environment variable expansion in configuration file.
 
-	-trace PATH
-	    Write verbose trace logging to PATH.
-
 `[1:], DefaultConfigPath())
 }

cmd/litestream/replicate_test.go

@@ -0,0 +1,136 @@
+package main_test
+
+import (
+	"context"
+	"database/sql"
+	"errors"
+	"fmt"
+	"hash/crc64"
+	"io"
+	"os"
+	"path/filepath"
+	"runtime"
+	"testing"
+	"time"
+
+	"golang.org/x/sync/errgroup"
+)
+
+func TestReplicateCommand(t *testing.T) {
+	if testing.Short() {
+		t.Skip("long running test, skipping")
+	} else if runtime.GOOS != "linux" {
+		t.Skip("must run system tests on Linux, skipping")
+	}
+
+	const writeTime = 10 * time.Second
+
+	dir := t.TempDir()
+	configPath := filepath.Join(dir, "litestream.yml")
+	dbPath := filepath.Join(dir, "db")
+	restorePath := filepath.Join(dir, "restored")
+	replicaPath := filepath.Join(dir, "replica")
+
+	if err := os.WriteFile(configPath, []byte(`
+dbs:
+  - path: `+dbPath+`
+    replicas:
+      - path: `+replicaPath+`
+`), 0666); err != nil {
+		t.Fatal(err)
+	}
+
+	// Generate data into SQLite database from separate goroutine.
+	g, ctx := errgroup.WithContext(context.Background())
+	mainctx, cancel := context.WithCancel(ctx)
+	g.Go(func() error {
+		defer cancel()
+
+		db, err := sql.Open("sqlite3", dbPath)
+		if err != nil {
+			return err
+		}
+		defer db.Close()
+
+		if _, err := db.ExecContext(ctx, `PRAGMA journal_mode = WAL`); err != nil {
+			return fmt.Errorf("cannot enable wal: %w", err)
+		} else if _, err := db.ExecContext(ctx, `PRAGMA synchronous = NORMAL`); err != nil {
+			return fmt.Errorf("cannot enable wal: %w", err)
+		} else if _, err := db.ExecContext(ctx, `CREATE TABLE t (id INTEGER PRIMARY KEY)`); err != nil {
+			return fmt.Errorf("cannot create table: %w", err)
+		}
+
+		ticker := time.NewTicker(1 * time.Millisecond)
+		defer ticker.Stop()
+		timer := time.NewTimer(writeTime)
+		defer timer.Stop()
+
+		for i := 0; ; i++ {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-timer.C:
+				return nil
+			case <-ticker.C:
+				if _, err := db.ExecContext(ctx, `INSERT INTO t (id) VALUES (?);`, i); err != nil {
+					return fmt.Errorf("cannot insert: i=%d err=%w", i, err)
+				}
+			}
+		}
+	})
+
+	// Replicate database unless the context is canceled.
+	g.Go(func() error {
+		m, _, _, _ := newMain()
+		return m.Run(mainctx, []string{"replicate", "-config", configPath})
+	})
+
+	if err := g.Wait(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Checkpoint database.
+	mustCheckpoint(t, dbPath)
+	chksum0 := mustChecksum(t, dbPath)
+
+	// Restore to another path.
+	m, _, _, _ := newMain()
+	if err := m.Run(context.Background(), []string{"restore", "-config", configPath, "-o", restorePath, dbPath}); err != nil && !errors.Is(err, context.Canceled) {
+		t.Fatal(err)
+	}
+
+	// Verify contents match.
+	if chksum1 := mustChecksum(t, restorePath); chksum0 != chksum1 {
+		t.Fatal("restore mismatch")
+	}
+}
+
+func mustCheckpoint(tb testing.TB, path string) {
+	tb.Helper()
+
+	db, err := sql.Open("sqlite3", path)
+	if err != nil {
+		tb.Fatal(err)
+	}
+	defer db.Close()
+
+	if _, err := db.Exec(`PRAGMA wal_checkpoint(TRUNCATE)`); err != nil {
+		tb.Fatal(err)
+	}
+}
+
+func mustChecksum(tb testing.TB, path string) uint64 {
+	tb.Helper()
+
+	f, err := os.Open(path)
+	if err != nil {
+		tb.Fatal(err)
+	}
+	defer f.Close()
+
+	h := crc64.New(crc64.MakeTable(crc64.ISO))
+	if _, err := io.Copy(h, f); err != nil {
+		tb.Fatal(err)
+	}
+	return h.Sum64()
+}

cmd/litestream/restore.go

@@ -2,36 +2,60 @@ package main
 
 import (
 	"context"
-	"errors"
 	"flag"
 	"fmt"
+	"io"
 	"log"
 	"os"
+	"path/filepath"
 	"strconv"
-	"time"
 
 	"github.com/benbjohnson/litestream"
 )
 
 // RestoreCommand represents a command to restore a database from a backup.
-type RestoreCommand struct{}
+type RestoreCommand struct {
+	stdin  io.Reader
+	stdout io.Writer
+	stderr io.Writer
+
+	snapshotIndex int // index of snapshot to start from
+
+	// CLI options
+	configPath      string // path to config file
+	noExpandEnv     bool   // if true, do not expand env variables in config
+	outputPath      string // path to restore database to
+	replicaName     string // optional, name of replica to restore from
+	generation      string // optional, generation to restore
+	targetIndex     int    // optional, last WAL index to replay
+	ifDBNotExists   bool   // if true, skips restore if output path already exists
+	ifReplicaExists bool   // if true, skips if no backups exist
+	opt             litestream.RestoreOptions
+}
+
+// NewRestoreCommand returns a new instance of RestoreCommand.
+func NewRestoreCommand(stdin io.Reader, stdout, stderr io.Writer) *RestoreCommand {
+	return &RestoreCommand{
+		stdin:  stdin,
+		stdout: stdout,
+		stderr: stderr,
+
+		targetIndex: -1,
+		opt:         litestream.NewRestoreOptions(),
+	}
+}
 
 // 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)
+	registerConfigFlag(fs, &c.configPath, &c.noExpandEnv)
-	fs.StringVar(&opt.OutputPath, "o", "", "output path")
+	fs.StringVar(&c.outputPath, "o", "", "output path")
-	fs.StringVar(&opt.ReplicaName, "replica", "", "replica name")
+	fs.StringVar(&c.replicaName, "replica", "", "replica name")
-	fs.StringVar(&opt.Generation, "generation", "", "generation name")
+	fs.StringVar(&c.generation, "generation", "", "generation name")
-	fs.Var((*indexVar)(&opt.Index), "index", "wal index")
+	fs.Var((*indexVar)(&c.targetIndex), "index", "wal index")
-	fs.IntVar(&opt.Parallelism, "parallelism", opt.Parallelism, "parallelism")
+	fs.IntVar(&c.opt.Parallelism, "parallelism", c.opt.Parallelism, "parallelism")
-	ifDBNotExists := fs.Bool("if-db-not-exists", false, "")
+	fs.BoolVar(&c.ifDBNotExists, "if-db-not-exists", false, "")
-	ifReplicaExists := fs.Bool("if-replica-exists", false, "")
+	fs.BoolVar(&c.ifReplicaExists, "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
@@ -40,87 +64,107 @@ func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
 	} else if fs.NArg() > 1 {
 		return fmt.Errorf("too many arguments")
 	}
+	pathOrURL := fs.Arg(0)
 
-	// Parse timestamp, if specified.
+	// Ensure a generation is specified if target index is specified.
-	if *timestampStr != "" {
+	if c.targetIndex != -1 && c.generation == "" {
-		if opt.Timestamp, err = time.Parse(time.RFC3339, *timestampStr); err != nil {
+		return fmt.Errorf("must specify -generation flag when using -index flag")
-			return errors.New("invalid -timestamp, must specify in ISO 8601 format (e.g. 2000-01-01T00:00:00Z)")
-		}
 	}
 
-	// Instantiate logger if verbose output is enabled.
+	// Default to original database path if output path not specified.
-	if *verbose {
+	if !isURL(pathOrURL) && c.outputPath == "" {
-		opt.Logger = log.New(os.Stderr, "", log.LstdFlags|log.Lmicroseconds)
+		c.outputPath = pathOrURL
 	}
 
-	// Determine replica & generation to restore from.
+	// Exit successfully if the output file already exists and flag is set.
-	var r *litestream.Replica
+	if _, err := os.Stat(c.outputPath); os.IsNotExist(err) {
-	if isURL(fs.Arg(0)) {
+		// file doesn't exist, continue
-		if *configPath != "" {
+	} else if err != nil {
-			return fmt.Errorf("cannot specify a replica URL and the -config flag")
+		return err
-		}
+	} else if err == nil {
-		if r, err = c.loadFromURL(ctx, fs.Arg(0), *ifDBNotExists, &opt); err == errSkipDBExists {
+		if c.ifDBNotExists {
-			fmt.Println("database already exists, skipping")
+			fmt.Fprintln(c.stdout, "database already exists, skipping")
 			return nil
+		}
+		return fmt.Errorf("output file already exists: %s", c.outputPath)
+	}
+
+	// Load configuration.
+	config, err := ReadConfigFile(c.configPath, !c.noExpandEnv)
+	if err != nil {
+		return err
+	}
+
+	// Build replica from either a URL or config.
+	r, err := c.loadReplica(ctx, config, pathOrURL)
+	if err != nil {
+		return err
+	}
+
+	// Determine latest generation if one is not specified.
+	if c.generation == "" {
+		if c.generation, err = litestream.FindLatestGeneration(ctx, r.Client()); err == litestream.ErrNoGeneration {
+			// Return an error if no matching targets found.
+			// If optional flag set, return success. Useful for automated recovery.
+			if c.ifReplicaExists {
+				fmt.Fprintln(c.stdout, "no matching backups found, skipping")
+				return nil
+			}
+			return fmt.Errorf("no matching backups found")
 		} else if err != nil {
-			return err
+			return fmt.Errorf("cannot determine latest generation: %w", err)
-		}
-	} else {
-		if *configPath == "" {
-			*configPath = DefaultConfigPath()
-		}
-		if r, err = c.loadFromConfig(ctx, fs.Arg(0), *configPath, !*noExpandEnv, *ifDBNotExists, &opt); err == errSkipDBExists {
-			fmt.Println("database already exists, skipping")
-			return nil
-		} else if err != nil {
-			return err
 		}
 	}
 
-	// Return an error if no matching targets found.
+	// Determine the maximum available index for the generation if one is not specified.
-	// If optional flag set, return success. Useful for automated recovery.
+	if c.targetIndex == -1 {
-	if opt.Generation == "" {
+		if c.targetIndex, err = litestream.FindMaxIndexByGeneration(ctx, r.Client(), c.generation); err != nil {
-		if *ifReplicaExists {
+			return fmt.Errorf("cannot determine latest index in generation %q: %w", c.generation, err)
-			fmt.Println("no matching backups found")
-			return nil
 		}
-		return fmt.Errorf("no matching backups found")
 	}
 
-	return r.Restore(ctx, opt)
+	// Find lastest snapshot that occurs before the index.
+	// TODO: Optionally allow -snapshot-index
+	if c.snapshotIndex, err = litestream.FindSnapshotForIndex(ctx, r.Client(), c.generation, c.targetIndex); err != nil {
+		return fmt.Errorf("cannot find snapshot index: %w", err)
+	}
+
+	// Create parent directory if it doesn't already exist.
+	if err := os.MkdirAll(filepath.Dir(c.outputPath), 0700); err != nil {
+		return fmt.Errorf("cannot create parent directory: %w", err)
+	}
+
+	c.opt.Logger = log.New(c.stdout, "", log.LstdFlags|log.Lmicroseconds)
+
+	return litestream.Restore(ctx, r.Client(), c.outputPath, c.generation, c.snapshotIndex, c.targetIndex, c.opt)
 }
 
-// loadFromURL creates a replica & updates the restore options from a replica URL.
+func (c *RestoreCommand) loadReplica(ctx context.Context, config Config, arg string) (*litestream.Replica, error) {
-func (c *RestoreCommand) loadFromURL(ctx context.Context, replicaURL string, ifDBNotExists bool, opt *litestream.RestoreOptions) (*litestream.Replica, error) {
+	if isURL(arg) {
-	if opt.OutputPath == "" {
+		return c.loadReplicaFromURL(ctx, config, arg)
-		return nil, fmt.Errorf("output path required")
 	}
+	return c.loadReplicaFromConfig(ctx, config, arg)
+}
 
-	// Exit successfully if the output file already exists.
+// loadReplicaFromURL creates a replica & updates the restore options from a replica URL.
-	if _, err := os.Stat(opt.OutputPath); !os.IsNotExist(err) && ifDBNotExists {
+func (c *RestoreCommand) loadReplicaFromURL(ctx context.Context, config Config, replicaURL string) (*litestream.Replica, error) {
-		return nil, errSkipDBExists
+	if c.replicaName != "" {
+		return nil, fmt.Errorf("cannot specify both the replica URL and the -replica flag")
+	} else if c.outputPath == "" {
+		return nil, fmt.Errorf("output path required when using a replica URL")
 	}
 
 	syncInterval := litestream.DefaultSyncInterval
-	r, err := NewReplicaFromConfig(&ReplicaConfig{
+	return NewReplicaFromConfig(&ReplicaConfig{
-		URL:          replicaURL,
+		URL:             replicaURL,
-		SyncInterval: &syncInterval,
+		AccessKeyID:     config.AccessKeyID,
+		SecretAccessKey: config.SecretAccessKey,
+		SyncInterval:    &syncInterval,
 	}, nil)
-	if err != nil {
-		return nil, err
-	}
-	opt.Generation, _, err = r.CalcRestoreTarget(ctx, *opt)
-	return r, err
 }
 
-// loadFromConfig returns a replica & updates the restore options from a DB reference.
+// loadReplicaFromConfig returns replicas based on the specific config path.
-func (c *RestoreCommand) loadFromConfig(ctx context.Context, dbPath, configPath string, expandEnv, ifDBNotExists bool, opt *litestream.RestoreOptions) (*litestream.Replica, error) {
+func (c *RestoreCommand) loadReplicaFromConfig(ctx context.Context, config Config, dbPath string) (_ *litestream.Replica, err error) {
-	// Load configuration.
-	config, err := ReadConfigFile(configPath, expandEnv)
-	if err != nil {
-		return nil, err
-	}
-
 	// Lookup database from configuration file by path.
 	if dbPath, err = expand(dbPath); err != nil {
 		return nil, err
@@ -132,31 +176,40 @@ func (c *RestoreCommand) loadFromConfig(ctx context.Context, dbPath, configPath
 	db, err := NewDBFromConfig(dbConfig)
 	if err != nil {
 		return nil, err
+	} else if len(db.Replicas) == 0 {
+		return nil, fmt.Errorf("database has no replicas: %s", dbPath)
 	}
 
-	// Restore into original database path if not specified.
+	// Filter by replica name if specified.
-	if opt.OutputPath == "" {
+	if c.replicaName != "" {
-		opt.OutputPath = dbPath
+		r := db.Replica(c.replicaName)
+		if r == nil {
+			return nil, fmt.Errorf("replica %q not found", c.replicaName)
+		}
+		return r, nil
 	}
 
-	// Exit successfully if the output file already exists.
+	// Choose only replica if only one available and no name is specified.
-	if _, err := os.Stat(opt.OutputPath); !os.IsNotExist(err) && ifDBNotExists {
+	if len(db.Replicas) == 1 {
-		return nil, errSkipDBExists
+		return db.Replicas[0], nil
 	}
 
-	// Determine the appropriate replica & generation to restore from,
+	// A replica must be specified when restoring a specific generation with multiple replicas.
-	r, generation, err := db.CalcRestoreTarget(ctx, *opt)
+	if c.generation != "" {
+		return nil, fmt.Errorf("must specify -replica flag when restoring from a specific generation")
+	}
+
+	// Determine latest replica to restore from.
+	r, err := litestream.LatestReplica(ctx, db.Replicas)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("cannot determine latest replica: %w", err)
 	}
-	opt.Generation = generation
-
 	return r, nil
 }
 
 // Usage prints the help screen to STDOUT.
 func (c *RestoreCommand) Usage() {
-	fmt.Printf(`
+	fmt.Fprintf(c.stdout, `
 The restore command recovers a database from a previous snapshot and WAL.
 
 Usage:
@@ -186,10 +239,6 @@ Arguments:
 	    Restore up to a specific hex-encoded WAL index (inclusive).
 	    Defaults to use the highest available index.
 
-	-timestamp TIMESTAMP
-	    Restore to a specific point-in-time.
-	    Defaults to use the latest available backup.
-
 	-o PATH
 	    Output path of the restored database.
 	    Defaults to original DB path.
@@ -213,9 +262,6 @@ Examples:
 	# Restore latest replica for database to original location.
 	$ litestream restore /path/to/db
 
-	# Restore replica for database to a given point in time.
-	$ litestream restore -timestamp 2020-01-01T00:00:00Z /path/to/db
-
 	# Restore latest replica for database to new /tmp directory
 	$ litestream restore -o /tmp/db /path/to/db
 
@@ -229,5 +275,3 @@ Examples:
 		DefaultConfigPath(),
 	)
 }
-
-var errSkipDBExists = errors.New("database already exists, skipping")

cmd/litestream/restore_test.go

@@ -0,0 +1,330 @@
+package main_test
+
+import (
+	"context"
+	"flag"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/benbjohnson/litestream/internal/testingutil"
+)
+
+func TestRestoreCommand(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "ok")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+		tempDir := t.TempDir()
+
+		m, _, stdout, stderr := newMain()
+		if err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "-o", filepath.Join(tempDir, "db"), filepath.Join(testDir, "db")}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stderr.String(), ""; got != want {
+			t.Fatalf("stderr=%q, want %q", got, want)
+		}
+
+		// STDOUT has timing info so we need to grep per line.
+		lines := strings.Split(stdout.String(), "\n")
+		for i, substr := range []string{
+			`restoring snapshot 0000000000000000/0000000000000000 to ` + filepath.Join(tempDir, "db.tmp"),
+			`applied wal 0000000000000000/0000000000000000 elapsed=`,
+			`applied wal 0000000000000000/0000000000000001 elapsed=`,
+			`applied wal 0000000000000000/0000000000000002 elapsed=`,
+			`renaming database from temporary location`,
+		} {
+			if !strings.Contains(lines[i], substr) {
+				t.Fatalf("stdout: unexpected line %d:\n%s", i+1, stdout)
+			}
+		}
+	})
+
+	t.Run("ReplicaName", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "replica-name")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+		tempDir := t.TempDir()
+
+		m, _, stdout, stderr := newMain()
+		if err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "-o", filepath.Join(tempDir, "db"), "-replica", "replica1", filepath.Join(testDir, "db")}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stderr.String(), ""; got != want {
+			t.Fatalf("stderr=%q, want %q", got, want)
+		}
+
+		// STDOUT has timing info so we need to grep per line.
+		lines := strings.Split(stdout.String(), "\n")
+		for i, substr := range []string{
+			`restoring snapshot 0000000000000001/0000000000000001 to ` + filepath.Join(tempDir, "db.tmp"),
+			`no wal files found, snapshot only`,
+			`renaming database from temporary location`,
+		} {
+			if !strings.Contains(lines[i], substr) {
+				t.Fatalf("stdout: unexpected line %d:\n%s", i+1, stdout)
+			}
+		}
+	})
+
+	t.Run("ReplicaURL", func(t *testing.T) {
+		testDir := filepath.Join(testingutil.Getwd(t), "testdata", "restore", "replica-url")
+		tempDir := t.TempDir()
+		replicaURL := "file://" + filepath.ToSlash(testDir) + "/replica"
+
+		m, _, stdout, stderr := newMain()
+		if err := m.Run(context.Background(), []string{"restore", "-o", filepath.Join(tempDir, "db"), replicaURL}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stderr.String(), ""; got != want {
+			t.Fatalf("stderr=%q, want %q", got, want)
+		}
+
+		lines := strings.Split(stdout.String(), "\n")
+		for i, substr := range []string{
+			`restoring snapshot 0000000000000000/0000000000000000 to ` + filepath.Join(tempDir, "db.tmp"),
+			`no wal files found, snapshot only`,
+			`renaming database from temporary location`,
+		} {
+			if !strings.Contains(lines[i], substr) {
+				t.Fatalf("stdout: unexpected line %d:\n%s", i+1, stdout)
+			}
+		}
+	})
+
+	t.Run("LatestReplica", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "latest-replica")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+		tempDir := t.TempDir()
+
+		m, _, stdout, stderr := newMain()
+		if err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "-o", filepath.Join(tempDir, "db"), filepath.Join(testDir, "db")}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stderr.String(), ""; got != want {
+			t.Fatalf("stderr=%q, want %q", got, want)
+		}
+
+		lines := strings.Split(stdout.String(), "\n")
+		for i, substr := range []string{
+			`restoring snapshot 0000000000000001/0000000000000000 to ` + filepath.Join(tempDir, "db.tmp"),
+			`no wal files found, snapshot only`,
+			`renaming database from temporary location`,
+		} {
+			if !strings.Contains(lines[i], substr) {
+				t.Fatalf("stdout: unexpected line %d:\n%s", i+1, stdout)
+			}
+		}
+	})
+
+	t.Run("IfDBNotExistsFlag", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "if-db-not-exists-flag")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, stdout, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "-if-db-not-exists", filepath.Join(testDir, "db")})
+		if err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("IfReplicaExists", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "if-replica-exists-flag")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, stdout, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "-if-replica-exists", filepath.Join(testDir, "db")})
+		if err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("ErrNoBackups", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "no-backups")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+		tempDir := t.TempDir()
+
+		m, _, stdout, stderr := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "-o", filepath.Join(tempDir, "db"), filepath.Join(testDir, "db")})
+		if err == nil || err.Error() != `no matching backups found` {
+			t.Fatalf("unexpected error: %s", err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		} else if got, want := stderr.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stderr"))); got != want {
+			t.Fatalf("stderr=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("ErrNoGeneration", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "no-generation")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), filepath.Join(testDir, "db")})
+		if err == nil || err.Error() != `no matching backups found` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrOutputPathExists", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "output-path-exists")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), filepath.Join(testDir, "db")})
+		if err == nil || err.Error() != `output file already exists: `+filepath.Join(testDir, "db") {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrDatabaseOrReplicaRequired", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore"})
+		if err == nil || err.Error() != `database path or replica URL required` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrTooManyArguments", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "abc", "123"})
+		if err == nil || err.Error() != `too many arguments` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrInvalidFlags", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-no-such-flag"})
+		if err == nil || err.Error() != `flag provided but not defined: -no-such-flag` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrIndexFlagOnly", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-index", "0", "/var/lib/db"})
+		if err == nil || err.Error() != `must specify -generation flag when using -index flag` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrConfigFileNotFound", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", "/no/such/file", "/var/lib/db"})
+		if err == nil || err.Error() != `config file not found: /no/such/file` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrInvalidConfig", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "invalid-config")
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "/var/lib/db"})
+		if err == nil || !strings.Contains(err.Error(), `replica path cannot be a url`) {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrMkdir", func(t *testing.T) {
+		tempDir := t.TempDir()
+		if err := os.Mkdir(filepath.Join(tempDir, "noperm"), 0000); err != nil {
+			t.Fatal(err)
+		}
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-o", filepath.Join(tempDir, "noperm", "subdir", "db"), "/var/lib/db"})
+		if err == nil || !strings.Contains(err.Error(), `permission denied`) {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrNoOutputPathWithReplicaURL", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "file://path/to/replica"})
+		if err == nil || err.Error() != `output path required when using a replica URL` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrReplicaNameWithReplicaURL", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-replica", "replica0", "file://path/to/replica"})
+		if err == nil || err.Error() != `cannot specify both the replica URL and the -replica flag` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrInvalidReplicaURL", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-o", filepath.Join(t.TempDir(), "db"), "xyz://xyz"})
+		if err == nil || err.Error() != `unknown replica type in config: "xyz"` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrDatabaseNotFound", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "database-not-found")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "/no/such/db"})
+		if err == nil || err.Error() != `database not found in config: /no/such/db` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrNoReplicas", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "no-replicas")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+		tempDir := t.TempDir()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "-o", filepath.Join(tempDir, "db"), filepath.Join(testDir, "db")})
+		if err == nil || err.Error() != `database has no replicas: `+filepath.Join(testingutil.Getwd(t), testDir, "db") {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrReplicaNotFound", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "replica-not-found")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+		tempDir := t.TempDir()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "-o", filepath.Join(tempDir, "db"), "-replica", "no_such_replica", filepath.Join(testDir, "db")})
+		if err == nil || err.Error() != `replica "no_such_replica" not found` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrGenerationWithNoReplicaName", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "generation-with-no-replica")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+		tempDir := t.TempDir()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "-o", filepath.Join(tempDir, "db"), "-generation", "0000000000000000", filepath.Join(testDir, "db")})
+		if err == nil || err.Error() != `must specify -replica flag when restoring from a specific generation` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrNoSnapshotsAvailable", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "restore", "no-snapshots")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+		tempDir := t.TempDir()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"restore", "-config", filepath.Join(testDir, "litestream.yml"), "-o", filepath.Join(tempDir, "db"), "-generation", "0000000000000000", filepath.Join(testDir, "db")})
+		if err == nil || err.Error() != `cannot determine latest index in generation "0000000000000000": no snapshots available` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("Usage", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		if err := m.Run(context.Background(), []string{"restore", "-h"}); err != flag.ErrHelp {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+}

cmd/litestream/snapshots.go

@@ -4,8 +4,9 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"io"
 	"log"
-	"os"
+	"sort"
 	"text/tabwriter"
 	"time"
 
@@ -13,95 +14,90 @@ import (
 )
 
 // SnapshotsCommand represents a command to list snapshots for a command.
-type SnapshotsCommand struct{}
+type SnapshotsCommand struct {
+	stdin  io.Reader
+	stdout io.Writer
+	stderr io.Writer
+
+	configPath  string
+	noExpandEnv bool
+
+	replicaName string
+}
+
+// NewSnapshotsCommand returns a new instance of SnapshotsCommand.
+func NewSnapshotsCommand(stdin io.Reader, stdout, stderr io.Writer) *SnapshotsCommand {
+	return &SnapshotsCommand{
+		stdin:  stdin,
+		stdout: stdout,
+		stderr: stderr,
+	}
+}
 
 // Run executes the command.
-func (c *SnapshotsCommand) Run(ctx context.Context, args []string) (err error) {
+func (c *SnapshotsCommand) Run(ctx context.Context, args []string) (ret error) {
 	fs := flag.NewFlagSet("litestream-snapshots", flag.ContinueOnError)
-	configPath, noExpandEnv := registerConfigFlag(fs)
+	registerConfigFlag(fs, &c.configPath, &c.noExpandEnv)
-	replicaName := fs.String("replica", "", "replica name")
+	fs.StringVar(&c.replicaName, "replica", "", "replica name")
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
 		return err
 	} else if fs.NArg() == 0 || fs.Arg(0) == "" {
-		return fmt.Errorf("database path required")
+		return fmt.Errorf("database path or replica URL required")
 	} else if fs.NArg() > 1 {
 		return fmt.Errorf("too many arguments")
 	}
 
-	var db *litestream.DB
+	// Load configuration.
-	var r *litestream.Replica
+	config, err := ReadConfigFile(c.configPath, !c.noExpandEnv)
-	if isURL(fs.Arg(0)) {
+	if err != nil {
-		if *configPath != "" {
+		return err
-			return fmt.Errorf("cannot specify a replica URL and the -config flag")
+	}
-		}
-		if r, err = NewReplicaFromConfig(&ReplicaConfig{URL: fs.Arg(0)}, nil); err != nil {
-			return err
-		}
-	} else {
-		if *configPath == "" {
-			*configPath = DefaultConfigPath()
-		}
 
-		// Load configuration.
+	// Determine list of replicas to pull snapshots from.
-		config, err := ReadConfigFile(*configPath, !*noExpandEnv)
+	replicas, _, err := loadReplicas(ctx, config, fs.Arg(0), c.replicaName)
+	if err != nil {
+		return err
+	}
+
+	// Build list of snapshot metadata with associated replica.
+	var infos []replicaSnapshotInfo
+	for _, r := range replicas {
+		a, err := r.Snapshots(ctx)
 		if err != nil {
-			return err
+			log.Printf("cannot determine snapshots: %s", err)
+			ret = errExit // signal error return without printing message
+			continue
 		}
-
+		for i := range a {
-		// Lookup database from configuration file by path.
+			infos = append(infos, replicaSnapshotInfo{SnapshotInfo: a[i], replicaName: r.Name()})
-		if path, err := expand(fs.Arg(0)); err != nil {
-			return err
-		} else if dbc := config.DBConfig(path); dbc == nil {
-			return fmt.Errorf("database not found in config: %s", path)
-		} else if db, err = NewDBFromConfig(dbc); err != nil {
-			return err
-		}
-
-		// Filter by replica, if specified.
-		if *replicaName != "" {
-			if r = db.Replica(*replicaName); r == nil {
-				return fmt.Errorf("replica %q not found for database %q", *replicaName, db.Path())
-			}
 		}
 	}
 
-	// Find snapshots by db or replica.
+	// Sort snapshots by creation time from newest to oldest.
-	var replicas []*litestream.Replica
+	sort.Slice(infos, func(i, j int) bool { return infos[i].CreatedAt.After(infos[j].CreatedAt) })
-	if r != nil {
-		replicas = []*litestream.Replica{r}
-	} else {
-		replicas = db.Replicas
-	}
 
 	// List all snapshots.
-	w := tabwriter.NewWriter(os.Stdout, 0, 8, 2, ' ', 0)
+	w := tabwriter.NewWriter(c.stdout, 0, 8, 2, ' ', 0)
 	defer w.Flush()
 
 	fmt.Fprintln(w, "replica\tgeneration\tindex\tsize\tcreated")
-	for _, r := range replicas {
+	for _, info := range infos {
-		infos, err := r.Snapshots(ctx)
+		fmt.Fprintf(w, "%s\t%s\t%s\t%d\t%s\n",
-		if err != nil {
+			info.replicaName,
-			log.Printf("cannot determine snapshots: %s", err)
+			info.Generation,
-			continue
+			litestream.FormatIndex(info.Index),
-		}
+			info.Size,
-		for _, info := range infos {
+			info.CreatedAt.Format(time.RFC3339),
-			fmt.Fprintf(w, "%s\t%s\t%d\t%d\t%s\n",
+		)
-				r.Name(),
-				info.Generation,
-				info.Index,
-				info.Size,
-				info.CreatedAt.Format(time.RFC3339),
-			)
-		}
 	}
 
-	return nil
+	return ret
 }
 
 // Usage prints the help screen to STDOUT.
 func (c *SnapshotsCommand) Usage() {
-	fmt.Printf(`
+	fmt.Fprintf(c.stdout, `
 The snapshots command lists all snapshots available for a database or replica.
 
 Usage:
@@ -137,3 +133,9 @@ Examples:
 		DefaultConfigPath(),
 	)
 }
+
+// replicaSnapshotInfo represents snapshot metadata with associated replica name.
+type replicaSnapshotInfo struct {
+	litestream.SnapshotInfo
+	replicaName string
+}

cmd/litestream/snapshots_test.go

@@ -0,0 +1,128 @@
+package main_test
+
+import (
+	"context"
+	"flag"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/benbjohnson/litestream/internal/testingutil"
+)
+
+func TestSnapshotsCommand(t *testing.T) {
+	t.Run("OK", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "snapshots", "ok")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, stdout, _ := newMain()
+		if err := m.Run(context.Background(), []string{"snapshots", "-config", filepath.Join(testDir, "litestream.yml"), filepath.Join(testDir, "db")}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("ReplicaName", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "snapshots", "replica-name")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, stdout, _ := newMain()
+		if err := m.Run(context.Background(), []string{"snapshots", "-config", filepath.Join(testDir, "litestream.yml"), "-replica", "replica1", filepath.Join(testDir, "db")}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("ReplicaURL", func(t *testing.T) {
+		testDir := filepath.Join(testingutil.Getwd(t), "testdata", "snapshots", "replica-url")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+		replicaURL := "file://" + filepath.ToSlash(testDir) + "/replica"
+
+		m, _, stdout, _ := newMain()
+		if err := m.Run(context.Background(), []string{"snapshots", replicaURL}); err != nil {
+			t.Fatal(err)
+		} else if got, want := stdout.String(), string(testingutil.ReadFile(t, filepath.Join(testDir, "stdout"))); got != want {
+			t.Fatalf("stdout=%q, want %q", got, want)
+		}
+	})
+
+	t.Run("ErrDatabaseOrReplicaRequired", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"snapshots"})
+		if err == nil || err.Error() != `database path or replica URL required` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrTooManyArguments", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"snapshots", "abc", "123"})
+		if err == nil || err.Error() != `too many arguments` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrInvalidFlags", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"snapshots", "-no-such-flag"})
+		if err == nil || err.Error() != `flag provided but not defined: -no-such-flag` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrConfigFileNotFound", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"snapshots", "-config", "/no/such/file", "/var/lib/db"})
+		if err == nil || err.Error() != `config file not found: /no/such/file` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrInvalidConfig", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "snapshots", "invalid-config")
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"snapshots", "-config", filepath.Join(testDir, "litestream.yml"), "/var/lib/db"})
+		if err == nil || !strings.Contains(err.Error(), `replica path cannot be a url`) {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrDatabaseNotFound", func(t *testing.T) {
+		testDir := filepath.Join("testdata", "snapshots", "database-not-found")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"snapshots", "-config", filepath.Join(testDir, "litestream.yml"), "/no/such/db"})
+		if err == nil || err.Error() != `database not found in config: /no/such/db` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrReplicaNotFound", func(t *testing.T) {
+		testDir := filepath.Join(testingutil.Getwd(t), "testdata", "snapshots", "replica-not-found")
+		defer testingutil.Setenv(t, "LITESTREAM_TESTDIR", testDir)()
+
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"snapshots", "-config", filepath.Join(testDir, "litestream.yml"), "-replica", "no_such_replica", filepath.Join(testDir, "db")})
+		if err == nil || err.Error() != `replica "no_such_replica" not found for database "`+filepath.Join(testDir, "db")+`"` {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("ErrInvalidReplicaURL", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		err := m.Run(context.Background(), []string{"snapshots", "xyz://xyz"})
+		if err == nil || !strings.Contains(err.Error(), `unknown replica type in config: "xyz"`) {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+
+	t.Run("Usage", func(t *testing.T) {
+		m, _, _, _ := newMain()
+		if err := m.Run(context.Background(), []string{"snapshots", "-h"}); err != flag.ErrHelp {
+			t.Fatalf("unexpected error: %s", err)
+		}
+	})
+}

cmd/litestream/testdata/Makefile

@@ -0,0 +1,13 @@
+.PHONY: default
+default:
+	make -C generations/ok
+	make -C generations/no-database
+	make -C generations/replica-name
+	make -C generations/replica-url
+	make -C restore/latest-replica
+	make -C snapshots/ok
+	make -C snapshots/replica-name
+	make -C snapshots/replica-url
+	make -C wal/ok
+	make -C wal/replica-name
+	make -C wal/replica-url

cmd/litestream/testdata/databases/invalid-config/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: /var/lib/db
+    replicas:
+      - path: s3://bkt/db

cmd/litestream/testdata/databases/no-databases/litestream.yml

@@ -0,0 +1 @@
+dbs:

cmd/litestream/testdata/databases/no-databases/stdout

@@ -0,0 +1 @@
+No databases found in config file.

cmd/litestream/testdata/databases/ok/litestream.yml

@@ -0,0 +1,7 @@
+dbs:
+  - path: /var/lib/db
+    replicas:
+      - path: /var/lib/replica
+      - url:  s3://mybkt/db
+
+  - path: /my/other/db

cmd/litestream/testdata/databases/ok/stdout

@@ -0,0 +1,3 @@
+path          replicas
+/var/lib/db   file,s3
+/my/other/db  

cmd/litestream/testdata/generations/database-not-found/litestream.yml

@@ -0,0 +1,2 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db

cmd/litestream/testdata/generations/invalid-config/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: /var/lib/db
+    replicas:
+      - path: s3://bkt/db

cmd/litestream/testdata/generations/no-database/Makefile

@@ -0,0 +1,4 @@
+.PHONY: default
+default:
+	TZ=UTC touch -ct 200001010000 replica/generations/0000000000000000/snapshots/0000000000000000.snapshot.lz4
+	TZ=UTC touch -ct 200001020000 replica/generations/0000000000000001/snapshots/0000000000000000.snapshot.lz4

cmd/litestream/testdata/generations/no-database/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica

cmd/litestream/testdata/generations/no-database/stdout

@@ -0,0 +1,3 @@
+name  generation        lag  start                 end
+file  0000000000000000  -    2000-01-01T00:00:00Z  2000-01-01T00:00:00Z
+file  0000000000000001  -    2000-01-02T00:00:00Z  2000-01-02T00:00:00Z

cmd/litestream/testdata/generations/ok/Makefile

@@ -0,0 +1,9 @@
+.PHONY: default
+default:
+	TZ=UTC touch -ct 200001030000 db
+	TZ=UTC touch -ct 200001010000 replica/generations/0000000000000000/snapshots/0000000000000000.snapshot.lz4
+	TZ=UTC touch -ct 200001020000 replica/generations/0000000000000000/snapshots/0000000000000001.snapshot.lz4
+	TZ=UTC touch -ct 200001010000 replica/generations/0000000000000000/wal/0000000000000000/0000000000000000.wal.lz4
+	TZ=UTC touch -ct 200001020000 replica/generations/0000000000000000/wal/0000000000000000/0000000000000001.wal.lz4
+	TZ=UTC touch -ct 200001030000 replica/generations/0000000000000000/wal/0000000000000001/0000000000000000.wal.lz4
+	TZ=UTC touch -ct 200001010000 replica/generations/0000000000000001/snapshots/0000000000000000.snapshot.lz4

cmd/litestream/testdata/generations/ok/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica

cmd/litestream/testdata/generations/ok/stdout

@@ -0,0 +1,3 @@
+name  generation        lag      start                 end
+file  0000000000000000  0s       2000-01-01T00:00:00Z  2000-01-03T00:00:00Z
+file  0000000000000001  48h0m0s  2000-01-01T00:00:00Z  2000-01-01T00:00:00Z

cmd/litestream/testdata/generations/replica-name/Makefile

@@ -0,0 +1,5 @@
+.PHONY: default
+default:
+	TZ=UTC touch -ct 200001030000 db
+	TZ=UTC touch -ct 200001010000 replica0/generations/0000000000000000/snapshots/0000000000000000.snapshot.lz4
+	TZ=UTC touch -ct 200001020000 replica1/generations/0000000000000001/snapshots/0000000000000000.snapshot.lz4

cmd/litestream/testdata/generations/replica-name/litestream.yml

@@ -0,0 +1,7 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - name: replica0
+        path: $LITESTREAM_TESTDIR/replica0
+      - name: replica1
+        path: $LITESTREAM_TESTDIR/replica1

cmd/litestream/testdata/generations/replica-name/stdout

@@ -0,0 +1,2 @@
+name      generation        lag      start                 end
+replica1  0000000000000001  24h0m0s  2000-01-02T00:00:00Z  2000-01-02T00:00:00Z

cmd/litestream/testdata/generations/replica-not-found/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - url: s3://bkt/db

cmd/litestream/testdata/generations/replica-url/Makefile

@@ -0,0 +1,9 @@
+.PHONY: default
+default:
+	TZ=UTC touch -ct 200001010000 replica/generations/0000000000000000/snapshots/0000000000000000.snapshot.lz4
+	TZ=UTC touch -ct 200001020000 replica/generations/0000000000000000/snapshots/0000000000000001.snapshot.lz4
+	TZ=UTC touch -ct 200001010000 replica/generations/0000000000000000/wal/0000000000000000/0000000000000000.wal.lz4
+	TZ=UTC touch -ct 200001020000 replica/generations/0000000000000000/wal/0000000000000000/0000000000000001.wal.lz4
+	TZ=UTC touch -ct 200001030000 replica/generations/0000000000000000/wal/0000000000000001/0000000000000000.wal.lz4
+	TZ=UTC touch -ct 200001020000 replica/generations/0000000000000001/snapshots/0000000000000000.snapshot.lz4
+

cmd/litestream/testdata/generations/replica-url/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica

cmd/litestream/testdata/generations/replica-url/stdout

@@ -0,0 +1,3 @@
+name  generation        lag  start                 end
+file  0000000000000000  -    2000-01-01T00:00:00Z  2000-01-03T00:00:00Z
+file  0000000000000001  -    2000-01-02T00:00:00Z  2000-01-02T00:00:00Z

cmd/litestream/testdata/restore/database-not-found/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica

cmd/litestream/testdata/restore/generation-with-no-replica/litestream.yml

@@ -0,0 +1,5 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica0
+      - path: $LITESTREAM_TESTDIR/replica1

cmd/litestream/testdata/restore/if-db-not-exists-flag/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica

cmd/litestream/testdata/restore/if-db-not-exists-flag/stdout

@@ -0,0 +1 @@
+database already exists, skipping

cmd/litestream/testdata/restore/if-replica-exists-flag/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica

cmd/litestream/testdata/restore/if-replica-exists-flag/stdout

@@ -0,0 +1 @@
+no matching backups found, skipping

cmd/litestream/testdata/restore/invalid-config/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: /var/lib/db
+    replicas:
+      - path: s3://bkt/db

cmd/litestream/testdata/restore/latest-replica/Makefile

@@ -0,0 +1,6 @@
+.PHONY: default
+default:
+	TZ=UTC touch -ct 200001010000 replica0/generations/0000000000000000/snapshots/0000000000000000.snapshot.lz4
+	TZ=UTC touch -ct 200001020000 replica1/generations/0000000000000002/snapshots/0000000000000000.snapshot.lz4
+	TZ=UTC touch -ct 200001030000 replica0/generations/0000000000000001/snapshots/0000000000000000.snapshot.lz4
+

cmd/litestream/testdata/restore/latest-replica/litestream.yml

@@ -0,0 +1,7 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - name: replica0
+        path: $LITESTREAM_TESTDIR/replica0
+      - name: replica1
+        path: $LITESTREAM_TESTDIR/replica1

cmd/litestream/testdata/restore/no-backups/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica

cmd/litestream/testdata/restore/no-generation/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica

cmd/litestream/testdata/restore/no-replicas/litestream.yml

@@ -0,0 +1,2 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db

cmd/litestream/testdata/restore/no-snapshots/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica

cmd/litestream/testdata/restore/ok/README

@@ -0,0 +1,36 @@
+To reproduce this testdata, run sqlite3 and execute:
+
+	PRAGMA journal_mode = WAL;
+	CREATE TABLE t (x);
+	INSERT INTO t (x) VALUES (1);
+	INSERT INTO t (x) VALUES (2);
+
+	sl3 split -o generations/0000000000000000/wal/0000000000000000 db-wal
+	cp db generations/0000000000000000/snapshots/0000000000000000.snapshot
+	lz4 -c --rm generations/0000000000000000/snapshots/0000000000000000.snapshot
+
+
+Then execute:
+
+	PRAGMA wal_checkpoint(TRUNCATE);
+	INSERT INTO t (x) VALUES (3);
+
+	sl3 split -o generations/0000000000000000/wal/0000000000000001 db-wal
+
+
+Then execute:
+
+	PRAGMA wal_checkpoint(TRUNCATE);
+	INSERT INTO t (x) VALUES (4);
+	INSERT INTO t (x) VALUES (5);
+
+	sl3 split -o generations/0000000000000000/wal/0000000000000002 db-wal
+
+
+Finally, obtain the final snapshot:
+
+	PRAGMA wal_checkpoint(TRUNCATE);
+
+	cp db 0000000000000002.db
+	rm db*
+

cmd/litestream/testdata/restore/ok/litestream.yml

@@ -0,0 +1,4 @@
+dbs:
+  - path: $LITESTREAM_TESTDIR/db
+    replicas:
+      - path: $LITESTREAM_TESTDIR/replica