Refactor shadow WAL to use segments

.github/CONTRIBUTING.md

@@ -1,18 +1,17 @@
-## Contribution Policy
+## Open-source, not open-contribution
 
-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.
+[Similar to SQLite](https://www.sqlite.org/copyright.html), Litestream is open
+source but closed to contributions. This keeps the code base free of proprietary
+or licensed code but it also helps me continue to maintain and build Litestream.
 
-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.
-
-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
+As the author of [BoltDB](https://github.com/boltdb/bolt), I found that
+accepting and maintaining third party patches contributed to my burn out and
+I eventually archived the project. Writing databases & low-level replication
+tools involves nuance and simple one line changes can have profound and
+unexpected changes in correctness and performance. Small contributions
+typically required hours of my time to properly test and validate them.
 
+I am grateful for community involvement, bug reports, & feature requests. I do
+not wish to come off as anything but welcoming, however, I've
+made the decision to keep this project closed to contributions for my own
+mental health and long term viability of the project.

.github/pull_request_template.md

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

.github/workflows/build_and_test.yml

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

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

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

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

@@ -1,51 +0,0 @@
-on:
-  release:
-    types:
-      - published
-  pull_request:
-    types:
-      - opened
-      - synchronize
-      - reopened
-    branches-ignore:
-      - "dependabot/**"
-
-name: Release (Docker)
-jobs:
-  docker:
-    runs-on: ubuntu-latest
-    env:
-      PLATFORMS: "${{ 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,16 +1,9 @@
 on:
   release:
     types:
-      - published
-  pull_request:
-    types:
-      - opened
-      - synchronize
-      - reopened
-    branches-ignore:
-      - "dependabot/**"
+      - created
 
-name: Release (Linux)
+name: release (linux)
 jobs:
   build:
     runs-on: ubuntu-18.04
@@ -19,53 +12,31 @@ jobs:
         include:
           - arch: amd64
             cc:   gcc
-
-          - arch:   amd64
-            cc:     gcc
-            static: 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.17'
+          go-version: '1.16'
+
+      - id: release
+        uses: bruceadams/get-release@v1.2.2
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
 
       - name: Install cross-compilers
         run: |
@@ -79,56 +50,32 @@ jobs:
 
       - name: Build litestream
         run: |
-          rm -rf dist && mkdir -p dist
-
+          rm -rf dist
+          mkdir -p dist
           cp etc/litestream.yml etc/litestream.service dist
-          cat etc/nfpm.yml | LITESTREAM_VERSION=${{ env.VERSION }} envsubst > dist/nfpm.yml
-
-          CGO_ENABLED=1 go build -ldflags "-s -w ${{ env.LDFLAGS }} -X 'main.Version=${{ env.VERSION }}'" -tags "${{ env.TAGS }}" -o dist/litestream ./cmd/litestream
+          cat etc/nfpm.yml | LITESTREAM_VERSION=${{ steps.release.outputs.tag_name }} envsubst > dist/nfpm.yml
+          CGO_ENABLED=1 go build -ldflags "-s -w -X 'main.Version=${{ steps.release.outputs.tag_name }}'" -o dist/litestream ./cmd/litestream
           
           cd dist
-          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-${{ 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: Get release
-        id: release
-        uses: bruceadams/get-release@v1.2.3
-        if: github.event_name == 'release'
-        env:
-          GITHUB_TOKEN: ${{ github.token }}
+          tar -czvf litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.tar.gz litestream
+          ../nfpm pkg --config nfpm.yml --packager deb --target litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.deb 
 
       - 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-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.tar.gz
-          asset_name: litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.tar.gz
+          asset_path: ./dist/litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.tar.gz
+          asset_name: litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.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-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.deb
-          asset_name: litestream-${{ env.VERSION }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}${{ env.SUFFIX }}.deb
+          asset_path: ./dist/litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.deb
+          asset_name: litestream-${{ steps.release.outputs.tag_name }}-${{ env.GOOS }}-${{ env.GOARCH }}${{ env.GOARM }}.deb
           asset_content_type: application/octet-stream

.github/workflows/release.linux_static.yml

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

.github/workflows/test.yml

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

Dockerfile

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

Makefile

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

README.md

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

abs/replica_client.go

@@ -190,6 +190,8 @@ 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

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

cmd/litestream/databases.go

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

cmd/litestream/databases_test.go

@@ -1,66 +0,0 @@
-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,116 +4,117 @@ import (
 	"context"
 	"flag"
 	"fmt"
-	"io"
+	"log"
 	"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 {
-	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,
-	}
-}
+type GenerationsCommand struct{}
 
 // Run executes the command.
-func (c *GenerationsCommand) Run(ctx context.Context, args []string) (ret error) {
+func (c *GenerationsCommand) Run(ctx context.Context, args []string) (err error) {
 	fs := flag.NewFlagSet("litestream-generations", flag.ContinueOnError)
-	registerConfigFlag(fs, &c.configPath, &c.noExpandEnv)
-	fs.StringVar(&c.replicaName, "replica", "", "replica name")
+	configPath, noExpandEnv := registerConfigFlag(fs)
+	replicaName := fs.String("replica", "", "replica name")
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
 		return err
-	} else if fs.Arg(0) == "" {
+	} else if fs.NArg() == 0 || 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
+	var r *litestream.Replica
+	dbUpdatedAt := time.Now()
+	if isURL(fs.Arg(0)) {
+		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(c.configPath, !c.noExpandEnv)
+		config, err := ReadConfigFile(*configPath, !*noExpandEnv)
 		if err != nil {
 			return err
 		}
 
-	replicas, db, err := loadReplicas(ctx, config, fs.Arg(0), c.replicaName)
-	if err != nil {
+		// 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.
-	var dbUpdatedAt time.Time
-	if db != nil {
-		if dbUpdatedAt, err = db.UpdatedAt(); err != nil && !os.IsNotExist(err) {
+		if dbUpdatedAt, err = db.UpdatedAt(); err != nil {
 			return err
 		}
 	}
 
+	var replicas []*litestream.Replica
+	if r != nil {
+		replicas = []*litestream.Replica{r}
+	} else {
+		replicas = db.Replicas
+	}
+
 	// List each generation.
-	w := tabwriter.NewWriter(c.stdout, 0, 8, 2, ' ', 0)
+	w := tabwriter.NewWriter(os.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 {
-			fmt.Fprintf(c.stderr, "%s: cannot list generations: %s", r.Name(), err)
-			ret = errExit // signal error return without printing message
+			log.Printf("%s: cannot list generations: %s", r.Name(), err)
 			continue
 		}
 
 		// Iterate over each generation for the replica.
 		for _, generation := range generations {
-			createdAt, updatedAt, err := litestream.GenerationTimeBounds(ctx, r.Client(), generation)
+			createdAt, updatedAt, err := r.GenerationTimeBounds(ctx, generation)
 			if err != nil {
-				fmt.Fprintf(c.stderr, "%s: cannot determine generation time bounds: %s", r.Name(), err)
-				ret = errExit // signal error return without printing message
+				log.Printf("%s: cannot determine generation time bounds: %s", r.Name(), err)
 				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,
-				lag,
+				truncateDuration(dbUpdatedAt.Sub(updatedAt)).String(),
 				createdAt.Format(time.RFC3339),
 				updatedAt.Format(time.RFC3339),
 			)
 		}
 	}
 
-	return ret
+	return nil
 }
 
 // Usage prints the help message to STDOUT.
 func (c *GenerationsCommand) Usage() {
-	fmt.Fprintf(c.stdout, `
+	fmt.Printf(`
 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.
@@ -140,3 +141,29 @@ 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

@@ -1,140 +0,0 @@
-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,7 +5,6 @@ import (
 	"errors"
 	"flag"
 	"fmt"
-	"io"
 	"io/ioutil"
 	"log"
 	"net/url"
@@ -17,13 +16,12 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
-	"syscall"
 	"time"
 
 	"github.com/benbjohnson/litestream"
 	"github.com/benbjohnson/litestream/abs"
-	"github.com/benbjohnson/litestream/gs"
-	"github.com/benbjohnson/litestream/http"
+	"github.com/benbjohnson/litestream/file"
+	"github.com/benbjohnson/litestream/gcs"
 	"github.com/benbjohnson/litestream/s3"
 	"github.com/benbjohnson/litestream/sftp"
 	_ "github.com/mattn/go-sqlite3"
@@ -35,15 +33,14 @@ var (
 	Version = "(development build)"
 )
 
-// errExit is a terminal error for indicating program should quit.
-var errExit = errors.New("exit")
+// errStop is a terminal error for indicating program should quit.
+var errStop = errors.New("stop")
 
 func main() {
 	log.SetFlags(0)
-	log.SetOutput(os.Stdout)
 
-	m := NewMain(os.Stdin, os.Stdout, os.Stderr)
-	if err := m.Run(context.Background(), os.Args[1:]); err == flag.ErrHelp || err == errExit {
+	m := NewMain()
+	if err := m.Run(context.Background(), os.Args[1:]); err == flag.ErrHelp || err == errStop {
 		os.Exit(1)
 	} else if err != nil {
 		log.Println(err)
@@ -52,23 +49,22 @@ func main() {
 }
 
 // Main represents the main program execution.
-type Main struct {
-	stdin  io.Reader
-	stdout io.Writer
-	stderr io.Writer
-}
+type Main struct{}
 
 // NewMain returns a new instance of Main.
-func NewMain(stdin io.Reader, stdout, stderr io.Writer) *Main {
-	return &Main{
-		stdin:  stdin,
-		stdout: stdout,
-		stderr: stderr,
-	}
+func NewMain() *Main {
+	return &Main{}
 }
 
 // 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()
 
@@ -80,20 +76,19 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 
 	switch cmd {
 	case "databases":
-		return NewDatabasesCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
+		return (&DatabasesCommand{}).Run(ctx, args)
 	case "generations":
-		return NewGenerationsCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
+		return (&GenerationsCommand{}).Run(ctx, args)
 	case "replicate":
-		c := NewReplicateCommand(m.stdin, m.stdout, m.stderr)
+		c := NewReplicateCommand()
 		if err := c.ParseFlags(ctx, args); err != nil {
 			return err
 		}
 
 		// Setup signal handler.
 		ctx, cancel := context.WithCancel(ctx)
-		defer cancel()
 		signalCh := make(chan os.Signal, 1)
-		signal.Notify(signalCh, syscall.SIGINT, syscall.SIGTERM)
+		signal.Notify(signalCh, notifySignals...)
 
 		if err := c.Run(ctx); err != nil {
 			return err
@@ -102,21 +97,21 @@ 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")
+			fmt.Println("context done, litestream shutting down")
 		case err = <-c.execCh:
 			cancel()
-			fmt.Fprintln(m.stdout, "subprocess exited, litestream shutting down")
+			fmt.Println("subprocess exited, litestream shutting down")
 		case sig := <-signalCh:
 			cancel()
-			fmt.Fprintln(m.stdout, "signal received, litestream shutting down")
+			fmt.Println("signal received, litestream shutting down")
 
 			if c.cmd != nil {
-				fmt.Fprintln(m.stdout, "sending signal to exec process")
+				fmt.Println("sending signal to exec process")
 				if err := c.cmd.Process.Signal(sig); err != nil {
 					return fmt.Errorf("cannot signal exec process: %w", err)
 				}
 
-				fmt.Fprintln(m.stdout, "waiting for exec process to close")
+				fmt.Println("waiting for exec process to close")
 				if err := <-c.execCh; err != nil && !strings.HasPrefix(err.Error(), "signal:") {
 					return fmt.Errorf("cannot wait for exec process: %w", err)
 				}
@@ -127,17 +122,17 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 		if e := c.Close(); e != nil && err == nil {
 			err = e
 		}
-		fmt.Fprintln(m.stdout, "litestream shut down")
+		fmt.Println("litestream shut down")
 		return err
 
 	case "restore":
-		return NewRestoreCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
+		return (&RestoreCommand{}).Run(ctx, args)
 	case "snapshots":
-		return NewSnapshotsCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
+		return (&SnapshotsCommand{}).Run(ctx, args)
 	case "version":
-		return NewVersionCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
+		return (&VersionCommand{}).Run(ctx, args)
 	case "wal":
-		return NewWALCommand(m.stdin, m.stdout, m.stderr).Run(ctx, args)
+		return (&WALCommand{}).Run(ctx, args)
 	default:
 		if cmd == "" || cmd == "help" || strings.HasPrefix(cmd, "-") {
 			m.Usage()
@@ -149,7 +144,7 @@ func (m *Main) Run(ctx context.Context, args []string) (err error) {
 
 // Usage prints the help screen to STDOUT.
 func (m *Main) Usage() {
-	fmt.Fprintln(m.stdout, `
+	fmt.Println(`
 litestream is a tool for replicating SQLite databases.
 
 Usage:
@@ -216,34 +211,7 @@ 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.
-// 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) {
+func ReadConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 	config := DefaultConfig()
 
 	// Expand filename, if necessary.
@@ -253,9 +221,10 @@ 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 err != nil {
+	if os.IsNotExist(err) {
+		return config, fmt.Errorf("config file not found: %s", filename)
+	} else if err != nil {
 		return config, err
 	}
 
@@ -284,12 +253,10 @@ func readConfigFile(filename string, expandEnv bool) (_ Config, err error) {
 // DBConfig represents the configuration for a single database.
 type DBConfig struct {
 	Path               string         `yaml:"path"`
-	Upstream             UpstreamConfig `yaml:"upstream"`
-	MonitorDelayInterval *time.Duration `yaml:"monitor-delay-interval"`
+	MonitorInterval    *time.Duration `yaml:"monitor-interval"`
 	CheckpointInterval *time.Duration `yaml:"checkpoint-interval"`
 	MinCheckpointPageN *int           `yaml:"min-checkpoint-page-count"`
 	MaxCheckpointPageN *int           `yaml:"max-checkpoint-page-count"`
-	ShadowRetentionN     *int           `yaml:"shadow-retention-count"`
 
 	Replicas []*ReplicaConfig `yaml:"replicas"`
 }
@@ -300,27 +267,13 @@ 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.MonitorDelayInterval != nil {
-		db.MonitorDelayInterval = *dbc.MonitorDelayInterval
+	if dbc.MonitorInterval != nil {
+		db.MonitorInterval = *dbc.MonitorInterval
 	}
 	if dbc.CheckpointInterval != nil {
 		db.CheckpointInterval = *dbc.CheckpointInterval
@@ -331,9 +284,6 @@ func NewDBFromConfigWithPath(dbc *DBConfig, path string) (*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 {
@@ -347,11 +297,6 @@ func NewDBFromConfigWithPath(dbc *DBConfig, path string) (*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"
@@ -391,35 +336,8 @@ 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, client)
+	r := litestream.NewReplica(db, c.Name)
 	if v := c.Retention; v != nil {
 		r.Retention = *v
 	}
@@ -436,11 +354,37 @@ 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 FileReplicaClient built from config.
-func newFileReplicaClientFromConfig(c *ReplicaConfig) (_ *litestream.FileReplicaClient, err error) {
+// newFileReplicaClientFromConfig returns a new instance of file.ReplicaClient built from config.
+func newFileReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *file.ReplicaClient, 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")
@@ -465,11 +409,13 @@ func newFileReplicaClientFromConfig(c *ReplicaConfig) (_ *litestream.FileReplica
 	}
 
 	// Instantiate replica and apply time fields, if set.
-	return litestream.NewFileReplicaClient(path), nil
+	client := file.NewReplicaClient(path)
+	client.Replica = r
+	return client, nil
 }
 
 // newS3ReplicaClientFromConfig returns a new instance of s3.ReplicaClient built from config.
-func newS3ReplicaClientFromConfig(c *ReplicaConfig) (_ *s3.ReplicaClient, err error) {
+func newS3ReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *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")
@@ -531,13 +477,13 @@ func newS3ReplicaClientFromConfig(c *ReplicaConfig) (_ *s3.ReplicaClient, err er
 	return client, nil
 }
 
-// newGSReplicaClientFromConfig returns a new instance of gs.ReplicaClient built from config.
-func newGSReplicaClientFromConfig(c *ReplicaConfig) (_ *gs.ReplicaClient, err error) {
+// newGCSReplicaClientFromConfig returns a new instance of gcs.ReplicaClient built from config.
+func newGCSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *gcs.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 gs replica")
+		return nil, fmt.Errorf("cannot specify url & path for gcs replica")
 	} else if c.URL != "" && c.Bucket != "" {
-		return nil, fmt.Errorf("cannot specify url & bucket for gs replica")
+		return nil, fmt.Errorf("cannot specify url & bucket for gcs replica")
 	}
 
 	bucket, path := c.Bucket, c.Path
@@ -560,18 +506,18 @@ func newGSReplicaClientFromConfig(c *ReplicaConfig) (_ *gs.ReplicaClient, err er
 
 	// Ensure required settings are set.
 	if bucket == "" {
-		return nil, fmt.Errorf("bucket required for gs replica")
+		return nil, fmt.Errorf("bucket required for gcs replica")
 	}
 
 	// Build replica.
-	client := gs.NewReplicaClient()
+	client := gcs.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) (_ *abs.ReplicaClient, err error) {
+func newABSReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *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")
@@ -614,7 +560,7 @@ func newABSReplicaClientFromConfig(c *ReplicaConfig) (_ *abs.ReplicaClient, err
 }
 
 // newSFTPReplicaClientFromConfig returns a new instance of sftp.ReplicaClient built from config.
-func newSFTPReplicaClientFromConfig(c *ReplicaConfig) (_ *sftp.ReplicaClient, err error) {
+func newSFTPReplicaClientFromConfig(c *ReplicaConfig, r *litestream.Replica) (_ *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")
@@ -720,12 +666,12 @@ func DefaultConfigPath() string {
 	if v := os.Getenv("LITESTREAM_CONFIG"); v != "" {
 		return v
 	}
-	return "/etc/litestream.yml"
+	return defaultConfigPath
 }
 
-func registerConfigFlag(fs *flag.FlagSet, configPath *string, noExpandEnv *bool) {
-	fs.StringVar(configPath, "config", "", "config path")
-	fs.BoolVar(noExpandEnv, "no-expand-env", false, "do not expand env vars in config")
+func registerConfigFlag(fs *flag.FlagSet) (configPath *string, noExpandEnv *bool) {
+	return fs.String("config", "", "config path"),
+		fs.Bool("no-expand-env", false, "do not expand env vars in config")
 }
 
 // expand returns an absolute path for s.
@@ -759,7 +705,7 @@ var _ flag.Value = (*indexVar)(nil)
 
 // String returns an 8-character hexadecimal value.
 func (v *indexVar) String() string {
-	return litestream.FormatIndex(int(*v))
+	return fmt.Sprintf("%08x", int(*v))
 }
 
 // Set parses s into an integer from a hexadecimal value.
@@ -771,45 +717,3 @@ 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

@@ -0,0 +1,21 @@
+// +build !windows
+
+package main
+
+import (
+	"context"
+	"os"
+	"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")
+}
+
+var notifySignals = []os.Signal{syscall.SIGINT, syscall.SIGTERM}

cmd/litestream/main_test.go

@@ -1,8 +1,6 @@
 package main_test
 
 import (
-	"bytes"
-	"io"
 	"io/ioutil"
 	"log"
 	"os"
@@ -11,7 +9,8 @@ import (
 
 	"github.com/benbjohnson/litestream"
 	main "github.com/benbjohnson/litestream/cmd/litestream"
-	"github.com/benbjohnson/litestream/gs"
+	"github.com/benbjohnson/litestream/file"
+	"github.com/benbjohnson/litestream/gcs"
 	"github.com/benbjohnson/litestream/s3"
 )
 
@@ -104,7 +103,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().(*litestream.FileReplicaClient); !ok {
+	} else if client, ok := r.Client.(*file.ReplicaClient); !ok {
 		t.Fatal("unexpected replica type")
 	} else if got, want := client.Path(), "/foo"; got != want {
 		t.Fatalf("Path=%s, want %s", got, want)
@@ -116,7 +115,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)
@@ -135,7 +134,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)
@@ -154,7 +153,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)
@@ -170,11 +169,11 @@ func TestNewS3ReplicaFromConfig(t *testing.T) {
 	})
 }
 
-func TestNewGSReplicaFromConfig(t *testing.T) {
-	r, err := main.NewReplicaFromConfig(&main.ReplicaConfig{URL: "gs://foo/bar"}, nil)
+func TestNewGCSReplicaFromConfig(t *testing.T) {
+	r, err := main.NewReplicaFromConfig(&main.ReplicaConfig{URL: "gcs://foo/bar"}, nil)
 	if err != nil {
 		t.Fatal(err)
-	} else if client, ok := r.Client().(*gs.ReplicaClient); !ok {
+	} else if client, ok := r.Client.(*gcs.ReplicaClient); !ok {
 		t.Fatal("unexpected replica type")
 	} else if got, want := client.Bucket, "foo"; got != want {
 		t.Fatalf("Bucket=%s, want %s", got, want)
@@ -182,17 +181,3 @@ func TestNewGSReplicaFromConfig(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

@@ -0,0 +1,107 @@
+// +build windows
+
+package main
+
+import (
+	"context"
+	"io"
+	"log"
+	"os"
+
+	"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))
+}
+
+var notifySignals = []os.Signal{os.Interrupt}

cmd/litestream/replicate.go

@@ -4,45 +4,36 @@ 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/gs"
-	"github.com/benbjohnson/litestream/http"
+	"github.com/benbjohnson/litestream/file"
+	"github.com/benbjohnson/litestream/gcs"
 	"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
 
-	server     *litestream.Server
-	httpServer *http.Server
+	// List of managed databases specified in the config.
+	DBs []*litestream.DB
 }
 
-// NewReplicateCommand returns a new instance of ReplicateCommand.
-func NewReplicateCommand(stdin io.Reader, stdout, stderr io.Writer) *ReplicateCommand {
+func NewReplicateCommand() *ReplicateCommand {
 	return &ReplicateCommand{
-		stdin:  stdin,
-		stdout: stdout,
-		stderr: stderr,
-
 		execCh: make(chan error),
 	}
 }
@@ -51,8 +42,7 @@ func NewReplicateCommand(stdin io.Reader, stdout, stderr io.Writer) *ReplicateCo
 func (c *ReplicateCommand) ParseFlags(ctx context.Context, args []string) (err error) {
 	fs := flag.NewFlagSet("litestream-replicate", flag.ContinueOnError)
 	execFlag := fs.String("exec", "", "execute subcommand")
-	addr := fs.String("addr", "", "HTTP bind address (host:port)")
-	registerConfigFlag(fs, &c.configPath, &c.noExpandEnv)
+	configPath, noExpandEnv := registerConfigFlag(fs)
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -62,7 +52,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 c.configPath != "" {
+		if *configPath != "" {
 			return fmt.Errorf("cannot specify a replica URL and the -config flag")
 		}
 
@@ -76,18 +66,15 @@ func (c *ReplicateCommand) ParseFlags(ctx context.Context, args []string) (err e
 		}
 		c.Config.DBs = []*DBConfig{dbConfig}
 	} else {
-		if c.configPath == "" {
-			c.configPath = DefaultConfigPath()
+		if *configPath == "" {
+			*configPath = DefaultConfigPath()
 		}
-		if c.Config, err = ReadConfigFile(c.configPath, !c.noExpandEnv); err != nil {
+		if c.Config, err = ReadConfigFile(*configPath, !*noExpandEnv); err != nil {
 			return err
 		}
 	}
 
-	// Override config with flags, if specified.
-	if *addr != "" {
-		c.Config.Addr = *addr
-	}
+	// Override config exec command, if specified.
 	if *execFlag != "" {
 		c.Config.Exec = *execFlag
 	}
@@ -105,35 +92,29 @@ 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 {
-		path, err := expand(dbConfig.Path)
+		db, err := NewDBFromConfig(dbConfig)
 		if err != nil {
 			return err
 		}
 
-		if err := c.server.Watch(path, func(path string) (*litestream.DB, error) {
-			return NewDBFromConfigWithPath(dbConfig, path)
-		}); err != nil {
+		// Open database & attach to program.
+		if err := db.Open(); err != nil {
 			return err
 		}
+		c.DBs = append(c.DBs, db)
 	}
 
 	// Notify user that initialization is done.
-	for _, db := range c.server.DBs() {
+	for _, db := range c.DBs {
 		log.Printf("initialized db: %s", db.Path())
 		for _, r := range db.Replicas {
-			switch client := r.Client().(type) {
-			case *litestream.FileReplicaClient:
+			switch client := r.Client.(type) {
+			case *file.ReplicaClient:
 				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 *gs.ReplicaClient:
+			case *gcs.ReplicaClient:
 				log.Printf("replicating to: name=%q type=%q bucket=%q path=%q sync-interval=%s", r.Name(), client.Type(), client.Bucket, client.Path, r.SyncInterval)
 			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)
@@ -145,13 +126,22 @@ func (c *ReplicateCommand) Run(ctx context.Context) (err error) {
 		}
 	}
 
-	// Serve HTTP if enabled.
+	// Serve metrics over HTTP if enabled.
 	if c.Config.Addr != "" {
-		c.httpServer = http.NewServer(c.server, c.Config.Addr)
-		if err := c.httpServer.Open(); err != nil {
-			return fmt.Errorf("cannot start http server: %w", err)
+		hostport := c.Config.Addr
+		if host, port, _ := net.SplitHostPort(c.Config.Addr); port == "" {
+			return fmt.Errorf("must specify port for bind address: %q", c.Config.Addr)
+		} 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.
@@ -161,14 +151,8 @@ 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 = env
+		c.cmd.Env = os.Environ()
 		c.cmd.Stdout = os.Stdout
 		c.cmd.Stderr = os.Stderr
 		if err := c.cmd.Start(); err != nil {
@@ -177,29 +161,25 @@ 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 the HTTP server & all open databases.
+// Close closes all open databases.
 func (c *ReplicateCommand) Close() (err error) {
-	if c.httpServer != nil {
-		if e := c.httpServer.Close(); e != nil && err == nil {
+	for _, db := range c.DBs {
+		if e := db.SoftClose(); e != nil {
+			log.Printf("error closing db: path=%s err=%s", db.Path(), e)
+			if err == nil {
 				err = e
 			}
 		}
-	if c.server != nil {
-		if e := c.server.Close(); e != nil && err == nil {
-			err = e
-		}
 	}
 	return err
 }
 
 // Usage prints the help screen to STDOUT.
 func (c *ReplicateCommand) Usage() {
-	fmt.Fprintf(c.stdout, `
+	fmt.Printf(`
 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
@@ -221,10 +201,6 @@ 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.
 

cmd/litestream/replicate_test.go

@@ -13,6 +13,7 @@ import (
 	"testing"
 	"time"
 
+	main "github.com/benbjohnson/litestream/cmd/litestream"
 	"golang.org/x/sync/errgroup"
 )
 
@@ -81,8 +82,7 @@ dbs:
 
 	// Replicate database unless the context is canceled.
 	g.Go(func() error {
-		m, _, _, _ := newMain()
-		return m.Run(mainctx, []string{"replicate", "-config", configPath})
+		return main.NewMain().Run(mainctx, []string{"replicate", "-config", configPath})
 	})
 
 	if err := g.Wait(); err != nil {
@@ -94,8 +94,7 @@ dbs:
 	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) {
+	if err := main.NewMain().Run(context.Background(), []string{"restore", "-config", configPath, "-o", restorePath, dbPath}); err != nil && !errors.Is(err, context.Canceled) {
 		t.Fatal(err)
 	}
 

cmd/litestream/restore.go

@@ -2,60 +2,36 @@ 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 {
-	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(),
-	}
-}
+type RestoreCommand struct{}
 
 // Run executes the command.
 func (c *RestoreCommand) Run(ctx context.Context, args []string) (err error) {
+	opt := litestream.NewRestoreOptions()
+	opt.Verbose = true
+
 	fs := flag.NewFlagSet("litestream-restore", flag.ContinueOnError)
-	registerConfigFlag(fs, &c.configPath, &c.noExpandEnv)
-	fs.StringVar(&c.outputPath, "o", "", "output path")
-	fs.StringVar(&c.replicaName, "replica", "", "replica name")
-	fs.StringVar(&c.generation, "generation", "", "generation name")
-	fs.Var((*indexVar)(&c.targetIndex), "index", "wal index")
-	fs.IntVar(&c.opt.Parallelism, "parallelism", c.opt.Parallelism, "parallelism")
-	fs.BoolVar(&c.ifDBNotExists, "if-db-not-exists", false, "")
-	fs.BoolVar(&c.ifReplicaExists, "if-replica-exists", false, "")
+	configPath, noExpandEnv := registerConfigFlag(fs)
+	fs.StringVar(&opt.OutputPath, "o", "", "output path")
+	fs.StringVar(&opt.ReplicaName, "replica", "", "replica name")
+	fs.StringVar(&opt.Generation, "generation", "", "generation name")
+	fs.Var((*indexVar)(&opt.Index), "index", "wal index")
+	fs.IntVar(&opt.Parallelism, "parallelism", opt.Parallelism, "parallelism")
+	ifDBNotExists := fs.Bool("if-db-not-exists", false, "")
+	ifReplicaExists := fs.Bool("if-replica-exists", false, "")
+	timestampStr := fs.String("timestamp", "", "timestamp")
+	verbose := fs.Bool("v", false, "verbose output")
 	fs.Usage = c.Usage
 	if err := fs.Parse(args); err != nil {
 		return err
@@ -64,107 +40,87 @@ 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)
 
-	// Ensure a generation is specified if target index is specified.
-	if c.targetIndex != -1 && c.generation == "" {
-		return fmt.Errorf("must specify -generation flag when using -index flag")
+	// Parse timestamp, if specified.
+	if *timestampStr != "" {
+		if opt.Timestamp, err = time.Parse(time.RFC3339, *timestampStr); err != nil {
+			return errors.New("invalid -timestamp, must specify in ISO 8601 format (e.g. 2000-01-01T00:00:00Z)")
+		}
 	}
 
-	// Default to original database path if output path not specified.
-	if !isURL(pathOrURL) && c.outputPath == "" {
-		c.outputPath = pathOrURL
+	// Instantiate logger if verbose output is enabled.
+	if *verbose {
+		opt.Logger = log.New(os.Stderr, "", log.LstdFlags|log.Lmicroseconds)
 	}
 
-	// Exit successfully if the output file already exists and flag is set.
-	if _, err := os.Stat(c.outputPath); os.IsNotExist(err) {
-		// file doesn't exist, continue
+	// Determine replica & generation to restore from.
+	var r *litestream.Replica
+	if isURL(fs.Arg(0)) {
+		if *configPath != "" {
+			return fmt.Errorf("cannot specify a replica URL and the -config flag")
+		}
+		if r, err = c.loadFromURL(ctx, fs.Arg(0), *ifDBNotExists, &opt); err == errSkipDBExists {
+			fmt.Println("database already exists, skipping")
+			return nil
 		} else if err != nil {
 			return err
-	} else if err == nil {
-		if c.ifDBNotExists {
-			fmt.Fprintln(c.stdout, "database already exists, skipping")
+		}
+	} 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
-		}
-		return fmt.Errorf("output file already exists: %s", c.outputPath)
-	}
-
-	// Load configuration.
-	config, err := ReadConfigFile(c.configPath, !c.noExpandEnv)
-	if err != nil {
+		} else 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")
+	if opt.Generation == "" {
+		if *ifReplicaExists {
+			fmt.Println("no matching backups found")
 			return nil
 		}
 		return fmt.Errorf("no matching backups found")
-		} else if err != nil {
-			return fmt.Errorf("cannot determine latest generation: %w", err)
-		}
 	}
 
-	// Determine the maximum available index for the generation if one is not specified.
-	if c.targetIndex == -1 {
-		if c.targetIndex, err = litestream.FindMaxIndexByGeneration(ctx, r.Client(), c.generation); err != nil {
-			return fmt.Errorf("cannot determine latest index in generation %q: %w", c.generation, err)
-		}
+	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)
+// loadFromURL creates a replica & updates the restore options from a replica URL.
+func (c *RestoreCommand) loadFromURL(ctx context.Context, replicaURL string, ifDBNotExists bool, opt *litestream.RestoreOptions) (*litestream.Replica, error) {
+	if opt.OutputPath == "" {
+		return nil, fmt.Errorf("output path required")
 	}
 
-	// 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)
-}
-
-func (c *RestoreCommand) loadReplica(ctx context.Context, config Config, arg string) (*litestream.Replica, error) {
-	if isURL(arg) {
-		return c.loadReplicaFromURL(ctx, config, arg)
-	}
-	return c.loadReplicaFromConfig(ctx, config, arg)
-}
-
-// loadReplicaFromURL creates a replica & updates the restore options from a replica URL.
-func (c *RestoreCommand) loadReplicaFromURL(ctx context.Context, config Config, replicaURL string) (*litestream.Replica, error) {
-	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")
+	// Exit successfully if the output file already exists.
+	if _, err := os.Stat(opt.OutputPath); !os.IsNotExist(err) && ifDBNotExists {
+		return nil, errSkipDBExists
 	}
 
 	syncInterval := litestream.DefaultSyncInterval
-	return NewReplicaFromConfig(&ReplicaConfig{
+	r, err := NewReplicaFromConfig(&ReplicaConfig{
 		URL:          replicaURL,
-		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.
+func (c *RestoreCommand) loadFromConfig(ctx context.Context, dbPath, configPath string, expandEnv, ifDBNotExists bool, opt *litestream.RestoreOptions) (*litestream.Replica, error) {
+	// Load configuration.
+	config, err := ReadConfigFile(configPath, expandEnv)
+	if err != nil {
+		return nil, err
 	}
 
-// loadReplicaFromConfig returns replicas based on the specific config path.
-func (c *RestoreCommand) loadReplicaFromConfig(ctx context.Context, config Config, dbPath string) (_ *litestream.Replica, err error) {
 	// Lookup database from configuration file by path.
 	if dbPath, err = expand(dbPath); err != nil {
 		return nil, err
@@ -176,40 +132,31 @@ func (c *RestoreCommand) loadReplicaFromConfig(ctx context.Context, config Confi
 	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)
 	}
 
-	// Filter by replica name if specified.
-	if c.replicaName != "" {
-		r := db.Replica(c.replicaName)
-		if r == nil {
-			return nil, fmt.Errorf("replica %q not found", c.replicaName)
-		}
-		return r, nil
+	// Restore into original database path if not specified.
+	if opt.OutputPath == "" {
+		opt.OutputPath = dbPath
 	}
 
-	// Choose only replica if only one available and no name is specified.
-	if len(db.Replicas) == 1 {
-		return db.Replicas[0], nil
+	// Exit successfully if the output file already exists.
+	if _, err := os.Stat(opt.OutputPath); !os.IsNotExist(err) && ifDBNotExists {
+		return nil, errSkipDBExists
 	}
 
-	// A replica must be specified when restoring a specific generation with multiple replicas.
-	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)
+	// Determine the appropriate replica & generation to restore from,
+	r, generation, err := db.CalcRestoreTarget(ctx, *opt)
 	if err != nil {
-		return nil, fmt.Errorf("cannot determine latest replica: %w", err)
+		return nil, err
 	}
+	opt.Generation = generation
+
 	return r, nil
 }
 
 // Usage prints the help screen to STDOUT.
 func (c *RestoreCommand) Usage() {
-	fmt.Fprintf(c.stdout, `
+	fmt.Printf(`
 The restore command recovers a database from a previous snapshot and WAL.
 
 Usage:
@@ -239,6 +186,10 @@ 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.
@@ -262,6 +213,9 @@ 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
 
@@ -275,3 +229,5 @@ Examples:
 		DefaultConfigPath(),
 	)
 }
+
+var errSkipDBExists = errors.New("database already exists, skipping")

cmd/litestream/restore_test.go

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

cmd/litestream/snapshots_test.go

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

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

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

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

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

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

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

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

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

cmd/litestream/testdata/databases/ok/stdout

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

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

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

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

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

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

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

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

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

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

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

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

cmd/litestream/testdata/generations/ok/stdout

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

cmd/litestream/testdata/restore/ok/README

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

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