docs: 0.1.1 changelog

refs #162

.circleci/config.yml

@@ -53,6 +53,13 @@ jobs:
       - run: sudo apt install python3 python3-pip libgirepository1.0-dev
       - run: ./lazy.sh devsetup
 
+      - run:
+          name: Install minio client, prepare minio
+          command: |
+            curl -sSL https://dl.minio.io/client/mc/release/linux-amd64/mc -o ${GOPATH}/bin/mc
+            chmod +x ${GOPATH}/bin/mc
+            mc config host add --api s3v4 zrepl-minio https://minio.cschwarz.com ${MINIO_ACCESS_KEY} ${MINIO_SECRET_KEY}
+
       - run: make vendordeps
       - save_cache:
           key: vendor
@@ -62,12 +69,28 @@ jobs:
       - run: make
       - run: make vet
       - run: make test
+      - run: make lint
       - run: make release
 
+
       - store_artifacts:
           path: ./artifacts/release
           when: always
 
+      - run:
+          shell: /bin/bash -euo pipefail
+          when: always
+          command: |
+            echo "$CIRCLE_BUILD_URL" > ./artifacts/release/cirlceci_build_url
+            mc cp -r artifacts/release "zrepl-minio/zrepl-ci-artifacts/${CIRCLE_SHA1}/${CIRCLE_JOB}/"
+            REPO="zrepl/zrepl"
+            COMMIT="${CIRCLE_SHA1}"
+            JOB_NAME="${CIRCLE_JOB}"
+            curl "https://api.github.com/repos/$REPO/statuses/$COMMIT?access_token=$GITHUB_COMMIT_STATUS_TOKEN" \
+              -H "Content-Type: application/json" \
+              -X POST \
+              -d '{"context":"zrepl/publish-ci-artifacts", "state": "success", "description":"CI Build Artifacts for '"$JOB_NAME"'", "target_url":"https://minio.cschwarz.com/minio/zrepl-ci-artifacts/'"$COMMIT"'/"}'
+
 
   build-1.11:
     <<: *build-latest

.golangci.yml

@@ -0,0 +1,10 @@
+linters:
+  enable:
+  - goimports
+
+issues:
+  exclude-rules:
+    - path: _test\.go
+      linters:
+        - errcheck
+

.travis.yml

@@ -29,6 +29,7 @@ matrix:
       - make
       - make vet
       - make test
+      - make lint
       - make artifacts/zrepl-freebsd-amd64
       - make artifacts/zrepl-linux-amd64
       - make artifacts/zrepl-darwin-amd64

Gopkg.lock

@@ -1,6 +1,22 @@
 # This file is autogenerated, do not edit; changes may be undone by the next 'dep ensure'.
 
 
+[[projects]]
+  digest = "1:e4b30804a381d7603b8a344009987c1ba351c26043501b23b8c7ce21f0b67474"
+  name = "github.com/BurntSushi/toml"
+  packages = ["."]
+  pruneopts = ""
+  revision = "3012a1dbe2e4bd1391d42b32f0577cb7bbc7f005"
+  version = "v0.3.1"
+
+[[projects]]
+  branch = "master"
+  digest = "1:bf241ce6eec44d9ebca17e22af43f13c25f55267b308ce8b29b9820c0d6d1d25"
+  name = "github.com/OpenPeeDeeP/depguard"
+  packages = ["."]
+  pruneopts = ""
+  revision = "1f388ab2d81096755d25043aa729e2fb889f3dae"
+
 [[projects]]
   branch = "master"
   digest = "1:8cf2cf1ab10480b5e0df950dac1517aaabde05d055d9d955652997ae4b9ecbbf"
@@ -33,6 +49,14 @@
   revision = "5b77d2a35fb0ede96d138fc9a99f5c9b6aef11b4"
   version = "v1.7.0"
 
+[[projects]]
+  digest = "1:eb53021a8aa3f599d29c7102e65026242bdedce998a54837dc67f14b6a97c5fd"
+  name = "github.com/fsnotify/fsnotify"
+  packages = ["."]
+  pruneopts = ""
+  revision = "c2828203cd70a50dcccfb2761f8b1f8ceef9a8e9"
+  version = "v1.4.7"
+
 [[projects]]
   branch = "master"
   digest = "1:5d0a2385edf4ba44f3b7b76bc0436ceb8f62bf55aa5d540a9eb9ec6c58d86809"
@@ -61,6 +85,28 @@
   revision = "de7e78efa4a71b3f36c7154989c529dbdf9ae623"
   version = "v1.1.0"
 
+[[projects]]
+  digest = "1:9d3f086381a257229b34fcd5690c3e5cadcb5f365cb35757536f3c51ccbb9049"
+  name = "github.com/go-critic/go-critic"
+  packages = [
+    "checkers",
+    "checkers/internal/lintutil",
+  ]
+  pruneopts = ""
+  revision = "d7b3038bc7a1c35a1d02fdd7cf4094f0f1a12001"
+  version = "v0.3.4"
+
+[[projects]]
+  digest = "1:50907242db0cb4c5d982ae213b995e9176b917edb269b645097af3289d9a15da"
+  name = "github.com/go-lintpack/lintpack"
+  packages = [
+    ".",
+    "astwalk",
+  ]
+  pruneopts = ""
+  revision = "80adc0715ac409128d0b7212719896ad8d3444b7"
+  version = "v0.5.2"
+
 [[projects]]
   digest = "1:6a4a01d58b227c4b6b11111b9f172ec5c17682b82724e58e6daf3f19f4faccd8"
   name = "github.com/go-logfmt/logfmt"
@@ -69,6 +115,95 @@
   revision = "390ab7935ee28ec6b286364bba9b4dd6410cb3d5"
   version = "v0.3.0"
 
+[[projects]]
+  digest = "1:1119997895278e1b27810308d8f802590e713bf0263b13be304ca9e086bd22a7"
+  name = "github.com/go-toolsmith/astcast"
+  packages = ["."]
+  pruneopts = ""
+  revision = "a6cb19f07b66b859a53f3f2be6e4c3bba892db7e"
+  version = "v1.0.0"
+
+[[projects]]
+  digest = "1:bcff57ad40d16a950986eb45dae40ed142d51c702e41fad2177518071bcc3d40"
+  name = "github.com/go-toolsmith/astcopy"
+  packages = ["."]
+  pruneopts = ""
+  revision = "245af3020944a15e09072c8ad3883c1451d1fdef"
+  version = "v1.0.0"
+
+[[projects]]
+  digest = "1:f6629a0ef3b819e34d4634c7540edf0cb392054ade814c4a9587ab6a23685def"
+  name = "github.com/go-toolsmith/astequal"
+  packages = ["."]
+  pruneopts = ""
+  revision = "dcb477bfacd6e00a13c6d63bfc73db28dd343160"
+  version = "v1.0.0"
+
+[[projects]]
+  digest = "1:05a2d85ca9e1164efa7cfb988d44bff221382658e567d958a78734bb9cccb758"
+  name = "github.com/go-toolsmith/astfmt"
+  packages = ["."]
+  pruneopts = ""
+  revision = "0d74c731079884bda287cf8df9ce7b92e688af8c"
+  version = "v1.0.0"
+
+[[projects]]
+  digest = "1:a6f3d4784ec69928a0a54521dd2536e6bacc73edf988edefc2820230d7be7703"
+  name = "github.com/go-toolsmith/astp"
+  packages = ["."]
+  pruneopts = ""
+  revision = "6373270dee65bfb0479f2acd16d4c8e9d5db13f8"
+  version = "v1.0.0"
+
+[[projects]]
+  digest = "1:9ba1aaf89cddc1cfe5d9a4a83d16ff8778369eaa358278a3fa2ef97847ebdb35"
+  name = "github.com/go-toolsmith/strparse"
+  packages = ["."]
+  pruneopts = ""
+  revision = "830b6daa1241714c12a9b9a4a56849fe2f93aedc"
+  version = "v1.0.0"
+
+[[projects]]
+  digest = "1:b827014e6963ac236f3698ce5ca1c85ad3ccfbf722186eab207836e1d7b9d615"
+  name = "github.com/go-toolsmith/typep"
+  packages = ["."]
+  pruneopts = ""
+  revision = "cab1745ffd84a567b524317c7f90e96755b18fcf"
+  version = "v1.0.0"
+
+[[projects]]
+  digest = "1:9ab1b1c637d7c8f49e39d8538a650d7eb2137b076790cff69d160823b505964c"
+  name = "github.com/gobwas/glob"
+  packages = [
+    ".",
+    "compiler",
+    "match",
+    "syntax",
+    "syntax/ast",
+    "syntax/lexer",
+    "util/runes",
+    "util/strings",
+  ]
+  pruneopts = ""
+  revision = "5ccd90ef52e1e632236f7326478d4faa74f99438"
+  version = "v0.2.3"
+
+[[projects]]
+  digest = "1:fd53b471edb4c28c7d297f617f4da0d33402755f58d6301e7ca1197ef0a90937"
+  name = "github.com/gogo/protobuf"
+  packages = ["proto"]
+  pruneopts = ""
+  revision = "ba06b47c162d49f2af050fb4c75bcbc86a159d5c"
+  version = "v1.2.1"
+
+[[projects]]
+  digest = "1:530233672f656641b365f8efb38ed9fba80e420baff2ce87633813ab3755ed6d"
+  name = "github.com/golang/mock"
+  packages = ["gomock"]
+  pruneopts = ""
+  revision = "51421b967af1f557f93a59e0057aaf15ca02e29c"
+  version = "v1.2.0"
+
 [[projects]]
   digest = "1:3dd078fda7500c341bc26cfbc6c6a34614f295a2457149fc1045cab767cbcf18"
   name = "github.com/golang/protobuf"
@@ -89,6 +224,213 @@
   revision = "aa810b61a9c79d51363740d207bb46cf8e620ed5"
   version = "v1.2.0"
 
+[[projects]]
+  branch = "master"
+  digest = "1:f6a3ed95affdd867195dec281bf1d328dddb37590463eba1f7b39b44a9829e3d"
+  name = "github.com/golangci/check"
+  packages = [
+    "cmd/structcheck",
+    "cmd/varcheck",
+  ]
+  pruneopts = ""
+  revision = "cfe4005ccda277a820149d44d6ededc400cc99a2"
+
+[[projects]]
+  branch = "master"
+  digest = "1:262000a2de14b1d0a802acb611e7ee69208b7e3e08f7d0e62226bb324f12e375"
+  name = "github.com/golangci/dupl"
+  packages = [
+    ".",
+    "job",
+    "printer",
+    "suffixtree",
+    "syntax",
+    "syntax/golang",
+  ]
+  pruneopts = ""
+  revision = "3e9179ac440a0386ac7cc9a085fc44397c6b9bbc"
+
+[[projects]]
+  branch = "master"
+  digest = "1:2298a8780ede449cb58108de23925ac2a14cca8ac151cfae45ea5992054d6cf2"
+  name = "github.com/golangci/errcheck"
+  packages = [
+    "golangci",
+    "internal/errcheck",
+  ]
+  pruneopts = ""
+  revision = "ef45e06d44b6e018d817c16c762d448990adc5e0"
+
+[[projects]]
+  branch = "master"
+  digest = "1:9b38ad496c9dabd1c820609c481f59c6c9597926c6125810af3d7a71bf2d649c"
+  name = "github.com/golangci/go-misc"
+  packages = ["deadcode"]
+  pruneopts = ""
+  revision = "927a3d87b613e9f6f0fb7ef8bb8de8b83c30a5a2"
+
+[[projects]]
+  branch = "master"
+  digest = "1:e4bbd53b867030ca8b2e0f6d7338cec2373baf14109858312daa51a144f4a091"
+  name = "github.com/golangci/go-tools"
+  packages = [
+    "arg",
+    "callgraph",
+    "callgraph/static",
+    "config",
+    "deprecated",
+    "functions",
+    "internal/sharedcheck",
+    "lint",
+    "lint/lintdsl",
+    "lint/lintutil",
+    "lint/lintutil/format",
+    "simple",
+    "ssa",
+    "ssa/ssautil",
+    "ssautil",
+    "staticcheck",
+    "staticcheck/vrp",
+    "stylecheck",
+    "unused",
+    "version",
+  ]
+  pruneopts = ""
+  revision = "35a9f45a5db090b0227d692d823151104cd695fa"
+
+[[projects]]
+  branch = "master"
+  digest = "1:bc3387ddcbdacf135af0a16b9e9ec6ac7bf5a1f822f679d8d29c2d97cfcce205"
+  name = "github.com/golangci/goconst"
+  packages = ["."]
+  pruneopts = ""
+  revision = "041c5f2b40f3dd334a4a6ee6a3f84ca3fc70680a"
+
+[[projects]]
+  branch = "master"
+  digest = "1:c5c9e52a4aaca585c1ce9c79f5ea31d74d03da39dfccda0b140f93d6a1be17b7"
+  name = "github.com/golangci/gocyclo"
+  packages = ["pkg/gocyclo"]
+  pruneopts = ""
+  revision = "0a533e8fa43d6605069e94f455bf9d79d4b8ea8c"
+
+[[projects]]
+  branch = "master"
+  digest = "1:edccfa947bd237dcd1bceef56d1670c22930831ca196ff04f0e2c4a8483bf97b"
+  name = "github.com/golangci/gofmt"
+  packages = [
+    "gofmt",
+    "goimports",
+  ]
+  pruneopts = ""
+  revision = "0b8337e80d98f7eec18e4504a4557b34423fd039"
+
+[[projects]]
+  digest = "1:0071a728673f03bd75b65863a37d4b1c5bb06ffc4a4416f1a8b6b90f36b2c5e3"
+  name = "github.com/golangci/golangci-lint"
+  packages = [
+    "cmd/golangci-lint",
+    "pkg/commands",
+    "pkg/config",
+    "pkg/exitcodes",
+    "pkg/fsutils",
+    "pkg/golinters",
+    "pkg/goutil",
+    "pkg/lint",
+    "pkg/lint/astcache",
+    "pkg/lint/linter",
+    "pkg/lint/lintersdb",
+    "pkg/logutils",
+    "pkg/packages",
+    "pkg/printers",
+    "pkg/report",
+    "pkg/result",
+    "pkg/result/processors",
+    "pkg/timeutils",
+  ]
+  pruneopts = ""
+  revision = "901cf25e20f86b7e9dc6f73eaba5afbd0cbdc257"
+  version = "v1.15.0"
+
+[[projects]]
+  branch = "master"
+  digest = "1:30c45dd735f55c7dbd0ea6040e3ccc35f867532b8e1919016c0565510392417a"
+  name = "github.com/golangci/gosec"
+  packages = [
+    ".",
+    "rules",
+  ]
+  pruneopts = ""
+  revision = "8afd9cbb6cfb34a3b4d4d5711bafdc6640ae892f"
+
+[[projects]]
+  branch = "master"
+  digest = "1:081d9ed8ba13ebbd4bd3e1f17cd703f77268416074588c38ce985d654b1fc0e1"
+  name = "github.com/golangci/govet"
+  packages = [
+    ".",
+    "lib/cfg",
+    "lib/whitelist",
+  ]
+  pruneopts = ""
+  revision = "44ddbe260190d79165f4150b828650780405d801"
+
+[[projects]]
+  branch = "master"
+  digest = "1:7da7fde58cf7cf5e19f6a1c77eb153945b28cf03bab227e0d831897b7070b546"
+  name = "github.com/golangci/ineffassign"
+  packages = ["."]
+  pruneopts = ""
+  revision = "2ee8f2867dde308c46d401d6d30f6c644094b167"
+
+[[projects]]
+  branch = "master"
+  digest = "1:8e50794fcb5f229576cd7eda5627a6c2f20341079f0c571077a7ab807c518da9"
+  name = "github.com/golangci/lint-1"
+  packages = ["."]
+  pruneopts = ""
+  revision = "d2cdd8c0821928c61cb0903441f8b35457a98a61"
+
+[[projects]]
+  branch = "master"
+  digest = "1:8665edfb3c5371fbac9820d127fa0d9aed813cc2349a27a7d16064dd89fed146"
+  name = "github.com/golangci/maligned"
+  packages = ["."]
+  pruneopts = ""
+  revision = "b1d89398deca2fd3f8578e5a9551e819bd01ca5f"
+
+[[projects]]
+  digest = "1:dbf28ceee27335219701dd4c6639c767eee31e2abb61485cdb1044587a04c077"
+  name = "github.com/golangci/misspell"
+  packages = ["."]
+  pruneopts = ""
+  revision = "b90dc15cfd220ecf8bbc9043ecb928cef381f011"
+  version = "v0.3.4"
+
+[[projects]]
+  branch = "master"
+  digest = "1:045c2735b360cbebf398a0e9312aeafebf08fd38f0d51cb2aa0f9420364c3cd1"
+  name = "github.com/golangci/prealloc"
+  packages = ["."]
+  pruneopts = ""
+  revision = "215b22d4de21190b80ce05e7d8466677c1aa3223"
+
+[[projects]]
+  branch = "master"
+  digest = "1:c23cf3c7078c3ba927492557a40c1ee1755734d4fff0e7fbe6d2f092604dae6d"
+  name = "github.com/golangci/revgrep"
+  packages = ["."]
+  pruneopts = ""
+  revision = "276a5c0a103935ee65af49afc254a65335bf1fcf"
+
+[[projects]]
+  branch = "master"
+  digest = "1:c553e7c7483f2d6db1e84a27a18df144ed4041792d7556916369f86ccf5409fe"
+  name = "github.com/golangci/unconvert"
+  packages = ["."]
+  pruneopts = ""
+  revision = "28b1c447d1f4a810737ee6ab40ea6c1d0ceae4ad"
+
 [[projects]]
   digest = "1:ad92aa49f34cbc3546063c7eb2cabb55ee2278b72842eda80e2a20a8a06a8d73"
   name = "github.com/google/uuid"
@@ -97,6 +439,25 @@
   revision = "0cd6bf5da1e1c83f8b45653022c74f71af0538a4"
   version = "v1.1.1"
 
+[[projects]]
+  digest = "1:d14365c51dd1d34d5c79833ec91413bfbb166be978724f15701e17080dc06dec"
+  name = "github.com/hashicorp/hcl"
+  packages = [
+    ".",
+    "hcl/ast",
+    "hcl/parser",
+    "hcl/printer",
+    "hcl/scanner",
+    "hcl/strconv",
+    "hcl/token",
+    "json/parser",
+    "json/scanner",
+    "json/token",
+  ]
+  pruneopts = ""
+  revision = "8cb6e5b959231cc1119e43259c4a608f9c51a241"
+  version = "v1.0.0"
+
 [[projects]]
   branch = "master"
   digest = "1:cb09475f771b9167fb9333629f5d6a7161572602ea040f1094602b0dc8709878"
@@ -105,6 +466,25 @@
   pruneopts = ""
   revision = "db4671f3a9b8df855e993f7c94ec5ef1ffb0a23b"
 
+[[projects]]
+  digest = "1:765270f95ea68ad2150f6143eb8b9c0c17b038a7e2255b46580674471af00e27"
+  name = "github.com/kisielk/gotool"
+  packages = [
+    ".",
+    "internal/load",
+  ]
+  pruneopts = ""
+  revision = "80517062f582ea3340cd4baf70e86d539ae7d84d"
+  version = "v1.0.0"
+
+[[projects]]
+  digest = "1:0f51cee70b0d254dbc93c22666ea2abf211af81c1701a96d04e2284b408621db"
+  name = "github.com/konsorten/go-windows-terminal-sequences"
+  packages = ["."]
+  pruneopts = ""
+  revision = "f55edac94c9bbba5d6182a4be46d86a2c9b5b50e"
+  version = "v1.0.2"
+
 [[projects]]
   branch = "master"
   digest = "1:1ed9eeebdf24aadfbca57eb50e6455bd1d2474525e0f0d4454de8c8e9bc7ee9a"
@@ -137,6 +517,14 @@
   revision = "345fbb3dbcdb252d9985ee899a84963c0fa24c82"
   version = "v1.0"
 
+[[projects]]
+  digest = "1:961dc3b1d11f969370533390fdf203813162980c858e1dabe827b60940c909a5"
+  name = "github.com/magiconair/properties"
+  packages = ["."]
+  pruneopts = ""
+  revision = "c2353362d570a7bfa228149c62842019201cfb71"
+  version = "v1.8.0"
+
 [[projects]]
   digest = "1:9ea83adf8e96d6304f394d40436f2eb44c1dc3250d223b74088cc253a6cd0a1c"
   name = "github.com/mattn/go-colorable"
@@ -169,6 +557,22 @@
   revision = "3247c84500bff8d9fb6d579d800f20b3e091582c"
   version = "v1.0.0"
 
+[[projects]]
+  digest = "1:6dbb0eb72090871f2e58d1e37973fe3cb8c0f45f49459398d3fc740cb30e13bd"
+  name = "github.com/mitchellh/go-homedir"
+  packages = ["."]
+  pruneopts = ""
+  revision = "af06845cf3004701891bf4fdb884bfe4920b3727"
+  version = "v1.1.0"
+
+[[projects]]
+  digest = "1:bcc46a0fbd9e933087bef394871256b5c60269575bb661935874729c65bbbf60"
+  name = "github.com/mitchellh/mapstructure"
+  packages = ["."]
+  pruneopts = ""
+  revision = "3536a929edddb9a5b34bd6861dc4a9647cb459fe"
+  version = "v1.1.2"
+
 [[projects]]
   digest = "1:4ff67dde814694496d7aa31be44b900f9717a10c8bc9136b13f49c8ef97f439a"
   name = "github.com/montanaflynn/stats"
@@ -177,6 +581,24 @@
   revision = "63fbb2597b7a13043b453a4b819945badb8f8926"
   version = "v0.5.0"
 
+[[projects]]
+  digest = "1:9da71b9d17d6231f1486dc62d81af3f9d34535703ba9e7a60a902433c3091e3b"
+  name = "github.com/nbutton23/zxcvbn-go"
+  packages = [
+    ".",
+    "adjacency",
+    "data",
+    "entropy",
+    "frequency",
+    "match",
+    "matching",
+    "scoring",
+    "utils/math",
+  ]
+  pruneopts = ""
+  revision = "eafdab6b0663b4b528c35975c8b0e78be6e25261"
+  version = "v0.1"
+
 [[projects]]
   branch = "master"
   digest = "1:f60ff065b58bd53e641112b38bbda9d2684deb828393c7ffb89c69a1ee301d17"
@@ -185,6 +607,14 @@
   pruneopts = ""
   revision = "0fd16699aae1833640fca52a937944c6f3b1d58c"
 
+[[projects]]
+  digest = "1:894aef961c056b6d85d12bac890bf60c44e99b46292888bfa66caf529f804457"
+  name = "github.com/pelletier/go-toml"
+  packages = ["."]
+  pruneopts = ""
+  revision = "c01d1270ff3e442a8a57cddc1c92dc1138598194"
+  version = "v1.2.0"
+
 [[projects]]
   digest = "1:7365acd48986e205ccb8652cc746f09c8b7876030d53710ea6ef7d0bd0dcd7ca"
   name = "github.com/pkg/errors"
@@ -269,6 +699,33 @@
   revision = "1744e2970ca51c86172c8190fadad617561ed6e7"
   version = "v1.0.0"
 
+[[projects]]
+  digest = "1:b73fe282e350b3ef2c71d8ff08e929e0b9670b1bb5b7fde1d3c1b4cd6e6dc8b1"
+  name = "github.com/sirupsen/logrus"
+  packages = ["."]
+  pruneopts = ""
+  revision = "dae0fa8d5b0c810a8ab733fbd5510c7cae84eca4"
+  version = "v1.4.0"
+
+[[projects]]
+  digest = "1:956f655c87b7255c6b1ae6c203ebb0af98cf2a13ef2507e34c9bf1c0332ac0f5"
+  name = "github.com/spf13/afero"
+  packages = [
+    ".",
+    "mem",
+  ]
+  pruneopts = ""
+  revision = "588a75ec4f32903aa5e39a2619ba6a4631e28424"
+  version = "v1.2.2"
+
+[[projects]]
+  digest = "1:ae3493c780092be9d576a1f746ab967293ec165e8473425631f06658b6212afc"
+  name = "github.com/spf13/cast"
+  packages = ["."]
+  pruneopts = ""
+  revision = "8c9545af88b134710ab1cd196795e7f2388358d7"
+  version = "v1.3.0"
+
 [[projects]]
   branch = "master"
   digest = "1:146327ce93be37e68bd3ff8541090d96da8cb3adc9e35d57570e9170a29f6bf6"
@@ -277,6 +734,14 @@
   pruneopts = ""
   revision = "b78744579491c1ceeaaa3b40205e56b0591b93a3"
 
+[[projects]]
+  digest = "1:cc15ae4fbdb02ce31f3392361a70ac041f4f02e0485de8ffac92bd8033e3d26e"
+  name = "github.com/spf13/jwalterweatherman"
+  packages = ["."]
+  pruneopts = ""
+  revision = "94f6ae3ed3bceceafa716478c5fbf8d29ca601a1"
+  version = "v1.1.0"
+
 [[projects]]
   digest = "1:261bc565833ef4f02121450d74eb88d5ae4bd74bfe5d0e862cddb8550ec35000"
   name = "github.com/spf13/pflag"
@@ -285,6 +750,14 @@
   revision = "e57e3eeb33f795204c1ca35f56c44f83227c6e66"
   version = "v1.0.0"
 
+[[projects]]
+  digest = "1:90fe60ab6f827e308b0c8cc1e11dce8ff1e96a927c8b171271a3cb04dd517606"
+  name = "github.com/spf13/viper"
+  packages = ["."]
+  pruneopts = ""
+  revision = "9e56dacc08fbbf8c9ee2dbc717553c758ce42bc9"
+  version = "v1.3.2"
+
 [[projects]]
   digest = "1:3926a4ec9a4ff1a072458451aa2d9b98acd059a45b38f7335d31e06c3d6a0159"
   name = "github.com/stretchr/testify"
@@ -331,6 +804,14 @@
   pruneopts = ""
   revision = "08227ad854131f7dfcdfb12579fb73dd8a38a03a"
 
+[[projects]]
+  branch = "master"
+  digest = "1:36ef1d8645934b1744cc7d8726e00d3dd9d8d84c18617bf7367a3a6d532f3370"
+  name = "golang.org/x/crypto"
+  packages = ["ssh/terminal"]
+  pruneopts = ""
+  revision = "a5d413f7728c81fb97d96a2b722368945f651e78"
+
 [[projects]]
   branch = "master"
   digest = "1:ea539c13b066dac72a940b62f37600a20ab8e88057397c78f3197c1a48475425"
@@ -347,11 +828,22 @@
   pruneopts = ""
   revision = "351d144fa1fc0bd934e2408202be0c29f25e35a0"
 
+[[projects]]
+  branch = "master"
+  digest = "1:0142c968b74c157abbb0220c05fa2bdde8a3a4509d6134b35ef75d5b58afb721"
+  name = "golang.org/x/sync"
+  packages = ["semaphore"]
+  pruneopts = ""
+  revision = "e225da77a7e68af35c70ccbf71af2b83e6acac3c"
+
 [[projects]]
   branch = "master"
   digest = "1:f358024b019f87eecaadcb098113a40852c94fe58ea670ef3c3e2d2c7bd93db1"
   name = "golang.org/x/sys"
-  packages = ["unix"]
+  packages = [
+    "unix",
+    "windows",
+  ]
   pruneopts = ""
   revision = "4ed8d59d0b35e1e29334a206d1b3f38b1e5dfb31"
 
@@ -375,6 +867,7 @@
     "unicode/cldr",
     "unicode/norm",
     "unicode/rangetable",
+    "width",
   ]
   pruneopts = ""
   revision = "f21a4dfb5e38f5895301dc265a8def02365cc3d0"
@@ -384,7 +877,24 @@
   branch = "master"
   digest = "1:4cd780b2ee42c8eac9c02bfb6e6b52dcbaef770774458c8938f5cbfb73a7b6d3"
   name = "golang.org/x/tools"
-  packages = ["cmd/stringer"]
+  packages = [
+    "cmd/goimports",
+    "cmd/stringer",
+    "go/ast/astutil",
+    "go/buildutil",
+    "go/gcexportdata",
+    "go/internal/cgo",
+    "go/internal/gcimporter",
+    "go/loader",
+    "go/packages",
+    "go/ssa",
+    "go/ssa/ssautil",
+    "go/types/typeutil",
+    "imports",
+    "internal/fastwalk",
+    "internal/gopathwalk",
+    "internal/semver",
+  ]
   pruneopts = ""
   revision = "d0ca3933b724e6be513276cc2edb34e10d667438"
 
@@ -436,6 +946,54 @@
   revision = "df014850f6dee74ba2fc94874043a9f3f75fbfd8"
   version = "v1.17.0"
 
+[[projects]]
+  digest = "1:cedccf16b71e86db87a24f8d4c70b0a855872eb967cb906a66b95de56aefbd0d"
+  name = "gopkg.in/yaml.v2"
+  packages = ["."]
+  pruneopts = ""
+  revision = "51d6538a90f86fe93ac480b35f37b2be17fef232"
+  version = "v2.2.2"
+
+[[projects]]
+  branch = "master"
+  digest = "1:2a6012038cdeb9851f1a71497544820e17ff2772b3cf799d24a76208cb9843b8"
+  name = "mvdan.cc/interfacer"
+  packages = ["check"]
+  pruneopts = ""
+  revision = "c20040233aedb03da82d460eca6130fcd91c629a"
+
+[[projects]]
+  branch = "master"
+  digest = "1:68e12be99c0d3355e04eecba6bc302876268a134a0eecd75258d8fefe44a94ed"
+  name = "mvdan.cc/lint"
+  packages = ["."]
+  pruneopts = ""
+  revision = "adc824a0674b99099789b6188a058d485eaf61c0"
+
+[[projects]]
+  branch = "master"
+  digest = "1:4af0788cd865cab3c8276462de56bad858e178199415241c2420e13e95d8594c"
+  name = "mvdan.cc/unparam"
+  packages = ["check"]
+  pruneopts = ""
+  revision = "1b9ccfa71afe53433971717161c9666adfc4d8c5"
+
+[[projects]]
+  digest = "1:0778809e0f18d0c0c05105a5c1e583d2253c5fd66fbd2b79b00e5f6439402491"
+  name = "sourcegraph.com/sourcegraph/go-diff"
+  packages = ["diff"]
+  pruneopts = ""
+  revision = "c613306ac97fb4807862c088149199f0dab8685a"
+  version = "v0.5.0"
+
+[[projects]]
+  digest = "1:ffc8cfc88692d5daab7abac1d989e9f7fc09727e42a945702f8f2d6d67f0fd6c"
+  name = "sourcegraph.com/sqs/pbtypes"
+  packages = ["."]
+  pruneopts = ""
+  revision = "688c2c2cb411327a50aae0f89119af9f38b0fc03"
+  version = "v1.0.0"
+
 [solve-meta]
   analyzer-name = "dep"
   analyzer-version = 1
@@ -446,6 +1004,7 @@
     "github.com/go-logfmt/logfmt",
     "github.com/golang/protobuf/proto",
     "github.com/golang/protobuf/protoc-gen-go",
+    "github.com/golangci/golangci-lint/cmd/golangci-lint",
     "github.com/google/uuid",
     "github.com/jinzhu/copier",
     "github.com/kr/pretty",
@@ -464,7 +1023,9 @@
     "github.com/yudai/gojsondiff/formatter",
     "github.com/zrepl/yaml-config",
     "golang.org/x/net/context",
+    "golang.org/x/sync/semaphore",
     "golang.org/x/sys/unix",
+    "golang.org/x/tools/cmd/goimports",
     "golang.org/x/tools/cmd/stringer",
     "google.golang.org/grpc",
     "google.golang.org/grpc/codes",

Gopkg.toml

@@ -5,6 +5,8 @@ ignored = [
 required = [
     "golang.org/x/tools/cmd/stringer",
     "github.com/alvaroloes/enumer",
+    "github.com/golangci/golangci-lint/cmd/golangci-lint",
+    "golang.org/x/tools/cmd/goimports",
 ]
 
 [[constraint]]

Makefile

@@ -1,4 +1,4 @@
-.PHONY: generate build test vet cover release docs docs-clean clean vendordeps
+.PHONY: generate build test vet cover release docs docs-clean clean vendordeps format lint
 .DEFAULT_GOAL := build
 
 ARTIFACTDIR := artifacts
@@ -30,6 +30,12 @@ generate: #not part of the build, must do that manually
 	protoc -I=replication/logic/pdu --go_out=plugins=grpc:replication/logic/pdu replication/logic/pdu/pdu.proto
 	go generate -x ./...
 
+format:
+	goimports -srcdir . -local 'github.com/zrepl/zrepl' -w $(shell find . -type f -name '*.go' -not -path "./vendor/*" -not -name '*.pb.go' -not -name '*_enumer.go')
+
+lint:
+	golangci-lint run ./...
+
 build:
 	@echo "INFO: In case of missing dependencies, run 'make vendordeps'"
 	$(GO_BUILD) -o "$(ARTIFACTDIR)/zrepl"
@@ -68,7 +74,7 @@ docs-clean:
 
 .PHONY: $(RELEASE_BINS)
 # TODO: two wildcards possible
-$(RELEASE_BINS): $(ARTIFACTDIR)/zrepl-%-amd64: generate $(ARTIFACTDIR) vet test
+$(RELEASE_BINS): $(ARTIFACTDIR)/zrepl-%-amd64: generate $(ARTIFACTDIR) vet test lint
 	@echo "INFO: In case of missing dependencies, run 'make vendordeps'"
 	GOOS=$* GOARCH=amd64   $(GO_BUILD) -o "$(ARTIFACTDIR)/zrepl-$*-amd64"
 

build/build.go

@@ -11,7 +11,8 @@
 package main
 
 import (
-	_ "fmt"
+	"fmt"
+
 	_ "github.com/alvaroloes/enumer"
 	_ "github.com/golang/protobuf/protoc-gen-go"
 	_ "golang.org/x/tools/cmd/stringer"

cli/cli.go

@@ -2,10 +2,12 @@ package cli
 
 import (
 	"fmt"
+	"os"
+
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
+
 	"github.com/zrepl/zrepl/config"
-	"os"
 )
 
 var rootArgs struct {
@@ -23,7 +25,10 @@ var bashcompCmd = &cobra.Command{
 	Run: func(cmd *cobra.Command, args []string) {
 		if len(args) != 1 {
 			fmt.Fprintf(os.Stderr, "specify exactly one positional agument\n")
-			cmd.Usage()
+			err := cmd.Usage()
+			if err != nil {
+				panic(err)
+			}
 			os.Exit(1)
 		}
 		if err := rootCmd.GenBashCompletionFile(args[0]); err != nil {
@@ -110,7 +115,6 @@ func addSubcommandToCobraCmd(c *cobra.Command, s *Subcommand) {
 	c.AddCommand(&cmd)
 }
 
-
 func Run() {
 	if err := rootCmd.Execute(); err != nil {
 		os.Exit(1)

client/configcheck.go

@@ -3,16 +3,18 @@ package client
 import (
 	"encoding/json"
 	"fmt"
+	"os"
+
 	"github.com/kr/pretty"
 	"github.com/pkg/errors"
 	"github.com/spf13/pflag"
 	"github.com/zrepl/yaml-config"
+
 	"github.com/zrepl/zrepl/cli"
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
-	"os"
 )
 
 var configcheckArgs struct {
@@ -28,14 +30,22 @@ var ConfigcheckCmd = &cli.Subcommand{
 		f.StringVar(&configcheckArgs.what, "what", "all", "what to print [all|config|jobs|logging]")
 	},
 	Run: func(subcommand *cli.Subcommand, args []string) error {
-		formatMap := map[string]func(interface{}) {
+		formatMap := map[string]func(interface{}){
 			"": func(i interface{}) {},
-			"pretty": func(i interface{}) { pretty.Println(i) },
+			"pretty": func(i interface{}) {
+				if _, err := pretty.Println(i); err != nil {
+					panic(err)
+				}
+			},
 			"json": func(i interface{}) {
-				json.NewEncoder(os.Stdout).Encode(subcommand.Config())
+				if err := json.NewEncoder(os.Stdout).Encode(subcommand.Config()); err != nil {
+					panic(err)
+				}
 			},
 			"yaml": func(i interface{}) {
-				yaml.NewEncoder(os.Stdout).Encode(subcommand.Config())
+				if err := yaml.NewEncoder(os.Stdout).Encode(subcommand.Config()); err != nil {
+					panic(err)
+				}
 			},
 		}
 
@@ -71,8 +81,7 @@ var ConfigcheckCmd = &cli.Subcommand{
 			}
 		}
 
-
+		whatMap := map[string]func(){
-		whatMap := map[string]func() {
 			"all": func() {
 				o := struct {
 					config  *config.Config
@@ -109,4 +118,3 @@ var ConfigcheckCmd = &cli.Subcommand{
 		}
 	},
 }
-

client/jsonclient.go

@@ -4,10 +4,11 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"github.com/pkg/errors"
 	"io"
 	"net"
 	"net/http"
+
+	"github.com/pkg/errors"
 )
 
 func controlHttpClient(sockpath string) (client http.Client, err error) {
@@ -35,7 +36,7 @@ func jsonRequestResponse(c http.Client, endpoint string, req interface{}, res in
 
 	if resp.StatusCode != http.StatusOK {
 		var msg bytes.Buffer
-		io.CopyN(&msg, resp.Body, 4096)
+		_, _ = io.CopyN(&msg, resp.Body, 4096) // ignore error, just display what we got
 		return errors.Errorf("%s", msg.String())
 	}
 

client/migrate.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/spf13/pflag"
+
 	"github.com/zrepl/zrepl/zfs"
 
 	"github.com/zrepl/zrepl/cli"
@@ -22,11 +23,6 @@ var (
 	}
 )
 
-type migration struct {
-	name   string
-	method func(config *config.Config, args []string) error
-}
-
 var migrations = []*cli.Subcommand{
 	&cli.Subcommand{
 		Use: "0.0.X:0.1:placeholder",

client/pprof.go

@@ -2,11 +2,12 @@ package client
 
 import (
 	"errors"
+	"log"
+	"os"
+
 	"github.com/zrepl/zrepl/cli"
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon"
-	"log"
-	"os"
 )
 
 var pprofArgs struct {

client/signal.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"github.com/pkg/errors"
+
 	"github.com/zrepl/zrepl/cli"
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon"

client/status.go

@@ -2,15 +2,6 @@ package client
 
 import (
 	"fmt"
-	"github.com/gdamore/tcell/termbox"
-	"github.com/pkg/errors"
-	"github.com/spf13/pflag"
-	"github.com/zrepl/yaml-config"
-	"github.com/zrepl/zrepl/cli"
-	"github.com/zrepl/zrepl/daemon"
-	"github.com/zrepl/zrepl/daemon/job"
-	"github.com/zrepl/zrepl/daemon/pruner"
-	"github.com/zrepl/zrepl/replication/report"
 	"io"
 	"math"
 	"net/http"
@@ -19,6 +10,17 @@ import (
 	"strings"
 	"sync"
 	"time"
+
+	"github.com/gdamore/tcell/termbox"
+	"github.com/pkg/errors"
+	"github.com/spf13/pflag"
+	"github.com/zrepl/yaml-config"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/daemon"
+	"github.com/zrepl/zrepl/daemon/job"
+	"github.com/zrepl/zrepl/daemon/pruner"
+	"github.com/zrepl/zrepl/replication/report"
 )
 
 type byteProgressMeasurement struct {
@@ -48,18 +50,17 @@ func (p *bytesProgressHistory) Update(currentVal int64) (bytesPerSecondAvg int64
 		p.lastChange = time.Now()
 	}
 
-	if time.Now().Sub(p.lastChange) > 3 * time.Second {
+	if time.Since(p.lastChange) > 3*time.Second {
 		p.last = nil
 		return 0, 0
 	}
 
-
+	deltaV := currentVal - p.last.val
-	deltaV := currentVal - p.last.val;
+	deltaT := time.Since(p.last.time)
-	deltaT := time.Now().Sub(p.last.time)
 	rate := float64(deltaV) / deltaT.Seconds()
 
 	factor := 0.3
-	p.bpsAvg =  (1-factor) * p.bpsAvg + factor * rate
+	p.bpsAvg = (1-factor)*p.bpsAvg + factor*rate
 
 	p.last.time = time.Now()
 	p.last.val = currentVal
@@ -80,15 +81,10 @@ type tui struct {
 
 func newTui() tui {
 	return tui{
-		replicationProgress: make(map[string]*bytesProgressHistory, 0),
+		replicationProgress: make(map[string]*bytesProgressHistory),
 	}
 }
 
-func (t *tui) moveCursor(x, y int) {
-	t.x += x
-	t.y += y
-}
-
 const INDENT_MULTIPLIER = 4
 
 func (t *tui) moveLine(dl int, col int) {
@@ -119,7 +115,7 @@ func wrap(s string, width int) string {
 			rem = len(s)
 		}
 		if idx := strings.IndexAny(s, "\n\r"); idx != -1 && idx < rem {
-			rem = idx+1
+			rem = idx + 1
 		}
 		untilNewline := strings.TrimRight(s[:rem], "\n\r")
 		s = s[rem:]
@@ -135,12 +131,12 @@ func wrap(s string, width int) string {
 func (t *tui) printfDrawIndentedAndWrappedIfMultiline(format string, a ...interface{}) {
 	whole := fmt.Sprintf(format, a...)
 	width, _ := termbox.Size()
-	if !strings.ContainsAny(whole, "\n\r") && t.x + len(whole) <= width {
+	if !strings.ContainsAny(whole, "\n\r") && t.x+len(whole) <= width {
 		t.printf(format, a...)
 	} else {
 		t.addIndent(1)
 		t.newline()
-		t.write(wrap(whole, width - INDENT_MULTIPLIER*t.indent))
+		t.write(wrap(whole, width-INDENT_MULTIPLIER*t.indent))
 		t.addIndent(-1)
 	}
 }
@@ -159,7 +155,6 @@ func (t *tui) addIndent(indent int) {
 	t.moveLine(0, 0)
 }
 
-
 var statusFlags struct {
 	Raw bool
 }
@@ -180,14 +175,17 @@ func runStatus(s *cli.Subcommand, args []string) error {
 	}
 
 	if statusFlags.Raw {
-		resp, err := httpc.Get("http://unix"+daemon.ControlJobEndpointStatus)
+		resp, err := httpc.Get("http://unix" + daemon.ControlJobEndpointStatus)
 		if err != nil {
 			return err
 		}
 		defer resp.Body.Close()
 		if resp.StatusCode != http.StatusOK {
 			fmt.Fprintf(os.Stderr, "Received error response:\n")
-			io.CopyN(os.Stderr, resp.Body, 4096)
+			_, err := io.CopyN(os.Stderr, resp.Body, 4096)
+			if err != nil {
+				return err
+			}
 			return errors.Errorf("exit")
 		}
 		if _, err := io.Copy(os.Stdout, resp.Body); err != nil {
@@ -226,7 +224,7 @@ func runStatus(s *cli.Subcommand, args []string) error {
 	ticker := time.NewTicker(500 * time.Millisecond)
 	defer ticker.Stop()
 	go func() {
-		for _ = range ticker.C {
+		for range ticker.C {
 			update()
 		}
 	}()
@@ -277,7 +275,7 @@ func (t *tui) draw() {
 		//Iterate over map in alphabetical order
 		keys := make([]string, len(t.report))
 		i := 0
-		for k, _ := range t.report {
+		for k := range t.report {
 			keys[i] = k
 			i++
 		}
@@ -363,7 +361,7 @@ func (t *tui) renderReplicationReport(rep *report.Report, history *bytesProgress
 		t.newline()
 	}
 	if !rep.WaitReconnectSince.IsZero() {
-		delta := rep.WaitReconnectUntil.Sub(time.Now()).Round(time.Second)
+		delta := time.Until(rep.WaitReconnectUntil).Round(time.Second)
 		if rep.WaitReconnectUntil.IsZero() || delta > 0 {
 			var until string
 			if rep.WaitReconnectUntil.IsZero() {
@@ -390,7 +388,7 @@ func (t *tui) renderReplicationReport(rep *report.Report, history *bytesProgress
 			t.newline()
 			t.addIndent(1)
 			for i, a := range rep.Attempts[:len(rep.Attempts)-1] {
-				t.printfDrawIndentedAndWrappedIfMultiline("#%d: %s (failed at %s) (ran %s)", i + 1, a.State, a.FinishAt, a.FinishAt.Sub(a.StartAt))
+				t.printfDrawIndentedAndWrappedIfMultiline("#%d: %s (failed at %s) (ran %s)", i+1, a.State, a.FinishAt, a.FinishAt.Sub(a.StartAt))
 				t.newline()
 			}
 			t.addIndent(-1)
@@ -462,7 +460,7 @@ func (t *tui) renderPrunerReport(r *pruner.Report) {
 		*pruner.FSReport
 		completed bool
 	}
-	all := make([]commonFS, 0, len(r.Pending) + len(r.Completed))
+	all := make([]commonFS, 0, len(r.Pending)+len(r.Completed))
 	for i := range r.Pending {
 		all = append(all, commonFS{&r.Pending[i], false})
 	}
@@ -471,7 +469,8 @@ func (t *tui) renderPrunerReport(r *pruner.Report) {
 	}
 
 	switch state {
-	case pruner.Plan: fallthrough
+	case pruner.Plan:
+		fallthrough
 	case pruner.PlanErr:
 		return
 	}
@@ -499,7 +498,7 @@ func (t *tui) renderPrunerReport(r *pruner.Report) {
 	t.write("[")
 	t.write(times("=", progress))
 	t.write(">")
-	t.write(times("-", 80 - progress))
+	t.write(times("-", 80-progress))
 	t.write("]")
 	t.printf(" %d/%d snapshots", completedDestroyCount, totalDestroyCount)
 	t.newline()
@@ -531,7 +530,7 @@ func (t *tui) renderPrunerReport(r *pruner.Report) {
 			len(fs.DestroyList), len(fs.SnapshotList))
 
 		if fs.completed {
-			t.printf( "Completed  %s\n", pruneRuleActionStr)
+			t.printf("Completed  %s\n", pruneRuleActionStr)
 			continue
 		}
 
@@ -560,14 +559,6 @@ func rightPad(str string, length int, pad string) string {
 	return str + times(pad, length-len(str))
 }
 
-
-func leftPad(str string, length int, pad string) string {
-	if len(str) > length {
-		return str[len(str)-length:]
-	}
-	return times(pad, length-len(str)) + str
-}
-
 var arrowPositions = `>\|/`
 
 // changeCount = 0 indicates stall / no progresss
@@ -584,7 +575,7 @@ func (t *tui) drawBar(length int, bytes, totalBytes int64, changeCount int) {
 
 	t.write("[")
 	t.write(times("=", completedLength))
-	t.write( string(arrowPositions[changeCount%len(arrowPositions)]))
+	t.write(string(arrowPositions[changeCount%len(arrowPositions)]))
 	t.write(times("-", length-completedLength))
 	t.write("]")
 }

client/stdinserver.go

@@ -1,13 +1,15 @@
 package client
 
 import (
-	"github.com/zrepl/zrepl/cli"
 	"os"
 
+	"github.com/problame/go-netssh"
+
+	"github.com/zrepl/zrepl/cli"
+	"github.com/zrepl/zrepl/config"
+
 	"context"
 	"errors"
-	"github.com/problame/go-netssh"
-	"github.com/zrepl/zrepl/config"
 	"log"
 	"path"
 )

client/testcmd.go

@@ -2,15 +2,17 @@ package client
 
 import (
 	"fmt"
+
 	"github.com/pkg/errors"
 	"github.com/spf13/pflag"
+
 	"github.com/zrepl/zrepl/cli"
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/zfs"
 )
 
-var TestCmd = &cli.Subcommand {
+var TestCmd = &cli.Subcommand{
 	Use: "test",
 	SetupSubcommands: func() []*cli.Subcommand {
 		return []*cli.Subcommand{testFilter, testPlaceholder}
@@ -51,8 +53,10 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 		return err
 	}
 	switch j := job.Ret.(type) {
-	case *config.SourceJob: confFilter = j.Filesystems
+	case *config.SourceJob:
-	case *config.PushJob: confFilter = j.Filesystems
+		confFilter = j.Filesystems
+	case *config.PushJob:
+		confFilter = j.Filesystems
 	default:
 		return fmt.Errorf("job type %T does not have filesystems filter", j)
 	}
@@ -109,9 +113,7 @@ func runTestFilterCmd(subcommand *cli.Subcommand, args []string) error {
 }
 
 var testPlaceholderArgs struct {
-	action string
 	ds  string
-	plv    string
 	all bool
 }
 

client/version.go

@@ -2,12 +2,14 @@ package client
 
 import (
 	"fmt"
+	"os"
+
 	"github.com/spf13/pflag"
+
 	"github.com/zrepl/zrepl/cli"
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon"
 	"github.com/zrepl/zrepl/version"
-	"os"
 )
 
 var versionArgs struct {

config/config.go

@@ -2,8 +2,6 @@ package config
 
 import (
 	"fmt"
-	"github.com/pkg/errors"
-	"github.com/zrepl/yaml-config"
 	"io/ioutil"
 	"log/syslog"
 	"os"
@@ -11,6 +9,9 @@ import (
 	"regexp"
 	"strconv"
 	"time"
+
+	"github.com/pkg/errors"
+	"github.com/zrepl/yaml-config"
 )
 
 type Config struct {
@@ -34,11 +35,16 @@ type JobEnum struct {
 func (j JobEnum) Name() string {
 	var name string
 	switch v := j.Ret.(type) {
-	case *SnapJob: name = v.Name
+	case *SnapJob:
-	case *PushJob: name = v.Name
+		name = v.Name
-	case *SinkJob: name = v.Name
+	case *PushJob:
-	case *PullJob: name = v.Name
+		name = v.Name
-	case *SourceJob: name = v.Name
+	case *SinkJob:
+		name = v.Name
+	case *PullJob:
+		name = v.Name
+	case *SourceJob:
+		name = v.Name
 	default:
 		panic(fmt.Sprintf("unknown job type %T", v))
 	}
@@ -402,7 +408,7 @@ func (t *ServeEnum) UnmarshalYAML(u func(interface{}, bool) error) (err error) {
 		"tcp":         &TCPServe{},
 		"tls":         &TLSServe{},
 		"stdinserver": &StdinserverServer{},
-		"local"      : &LocalServe{},
+		"local":       &LocalServe{},
 	})
 	return
 }
@@ -448,26 +454,46 @@ func (t *SyslogFacility) UnmarshalYAML(u func(interface{}, bool) error) (err err
 	}
 	var level syslog.Priority
 	switch s {
-		case "kern":     level = syslog.LOG_KERN
+	case "kern":
-		case "user":     level = syslog.LOG_USER
+		level = syslog.LOG_KERN
-		case "mail":     level = syslog.LOG_MAIL
+	case "user":
-		case "daemon":   level = syslog.LOG_DAEMON
+		level = syslog.LOG_USER
-		case "auth":     level = syslog.LOG_AUTH
+	case "mail":
-		case "syslog":   level = syslog.LOG_SYSLOG
+		level = syslog.LOG_MAIL
-		case "lpr":      level = syslog.LOG_LPR
+	case "daemon":
-		case "news":     level = syslog.LOG_NEWS
+		level = syslog.LOG_DAEMON
-		case "uucp":     level = syslog.LOG_UUCP
+	case "auth":
-		case "cron":     level = syslog.LOG_CRON
+		level = syslog.LOG_AUTH
-		case "authpriv": level = syslog.LOG_AUTHPRIV
+	case "syslog":
-		case "ftp":      level = syslog.LOG_FTP
+		level = syslog.LOG_SYSLOG
-		case "local0":   level = syslog.LOG_LOCAL0
+	case "lpr":
-		case "local1":   level = syslog.LOG_LOCAL1
+		level = syslog.LOG_LPR
-		case "local2":   level = syslog.LOG_LOCAL2
+	case "news":
-		case "local3":   level = syslog.LOG_LOCAL3
+		level = syslog.LOG_NEWS
-		case "local4":   level = syslog.LOG_LOCAL4
+	case "uucp":
-		case "local5":   level = syslog.LOG_LOCAL5
+		level = syslog.LOG_UUCP
-		case "local6":   level = syslog.LOG_LOCAL6
+	case "cron":
-		case "local7":   level = syslog.LOG_LOCAL7
+		level = syslog.LOG_CRON
+	case "authpriv":
+		level = syslog.LOG_AUTHPRIV
+	case "ftp":
+		level = syslog.LOG_FTP
+	case "local0":
+		level = syslog.LOG_LOCAL0
+	case "local1":
+		level = syslog.LOG_LOCAL1
+	case "local2":
+		level = syslog.LOG_LOCAL2
+	case "local3":
+		level = syslog.LOG_LOCAL3
+	case "local4":
+		level = syslog.LOG_LOCAL4
+	case "local5":
+		level = syslog.LOG_LOCAL5
+	case "local6":
+		level = syslog.LOG_LOCAL6
+	case "local7":
+		level = syslog.LOG_LOCAL7
 	default:
 		return fmt.Errorf("invalid syslog level: %q", s)
 	}

config/config_global_test.go

@@ -2,11 +2,12 @@ package config
 
 import (
 	"fmt"
+	"log/syslog"
+	"testing"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/zrepl/yaml-config"
-	"log/syslog"
-	"testing"
 )
 
 func testValidGlobalSection(t *testing.T, s string) *Config {
@@ -24,7 +25,7 @@ jobs:
 `
 	_, err := ParseConfigBytes([]byte(jobdef))
 	require.NoError(t, err)
-	return testValidConfig(t, s + jobdef)
+	return testValidConfig(t, s+jobdef)
 }
 
 func TestOutletTypes(t *testing.T) {

config/config_minimal_test.go

@@ -2,6 +2,7 @@ package config
 
 import (
 	"testing"
+
 	"github.com/stretchr/testify/assert"
 )
 

config/config_snapshotting_test.go

@@ -2,9 +2,10 @@ package config
 
 import (
 	"fmt"
-	"github.com/stretchr/testify/assert"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestSnapshotting(t *testing.T) {
@@ -37,7 +38,7 @@ jobs:
     interval: 10m
 `
 
-	fillSnapshotting := func(s string) string {return fmt.Sprintf(tmpl, s)}
+	fillSnapshotting := func(s string) string { return fmt.Sprintf(tmpl, s) }
 	var c *Config
 
 	t.Run("manual", func(t *testing.T) {
@@ -51,7 +52,7 @@ jobs:
 		snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic)
 		assert.Equal(t, "periodic", snp.Type)
 		assert.Equal(t, 10*time.Minute, snp.Interval)
-		assert.Equal(t, "zrepl_" , snp.Prefix)
+		assert.Equal(t, "zrepl_", snp.Prefix)
 	})
 
 }

config/config_test.go

@@ -39,6 +39,7 @@ func TestSampleConfigsAreParsedWithoutErrors(t *testing.T) {
 }
 
 // template must be a template/text template with a single '{{ . }}' as placehodler for val
+//nolint[:deadcode,unused]
 func testValidConfigTemplate(t *testing.T, tmpl string, val string) *Config {
 	tmp, err := template.New("master").Parse(tmpl)
 	if err != nil {

config/retentiongrid.go

@@ -31,10 +31,6 @@ func (i *RetentionInterval) KeepCount() int {
 
 const RetentionGridKeepCountAll int = -1
 
-type RetentionGrid struct {
-	intervals []RetentionInterval
-}
-
 func (t *RetentionIntervalList) UnmarshalYAML(u func(interface{}, bool) error) (err error) {
 	var in string
 	if err := u(&in, true); err != nil {

daemon/control.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/zrepl/zrepl/daemon/job"
 	"github.com/zrepl/zrepl/daemon/nethelpers"
 	"github.com/zrepl/zrepl/logger"
@@ -60,7 +61,7 @@ func (j *controlJob) RegisterMetrics(registerer prometheus.Registerer) {
 		Subsystem: "control",
 		Name:      "request_finished",
 		Help:      "time it took a request to finih",
-		Buckets: []float64{1e-6, 10e-6, 100e-6, 500e-6, 1e-3,10e-3, 100e-3, 200e-3,400e-3,800e-3, 1, 10, 20},
+		Buckets:   []float64{1e-6, 10e-6, 100e-6, 500e-6, 1e-3, 10e-3, 100e-3, 200e-3, 400e-3, 800e-3, 1, 10, 20},
 	}, []string{"endpoint"})
 	registerer.MustRegister(promControl.requestBegin)
 	registerer.MustRegister(promControl.requestFinished)
@@ -88,7 +89,7 @@ func (j *controlJob) Run(ctx context.Context) {
 
 	mux := http.NewServeMux()
 	mux.Handle(ControlJobEndpointPProf,
-		requestLogger{log: log, handler: jsonRequestResponder{func(decoder jsonDecoder) (interface{}, error) {
+		requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (interface{}, error) {
 			var msg PprofServerControlMsg
 			err := decoder(&msg)
 			if err != nil {
@@ -99,19 +100,19 @@ func (j *controlJob) Run(ctx context.Context) {
 		}}})
 
 	mux.Handle(ControlJobEndpointVersion,
-		requestLogger{log: log, handler: jsonResponder{func() (interface{}, error) {
+		requestLogger{log: log, handler: jsonResponder{log, func() (interface{}, error) {
 			return version.NewZreplVersionInformation(), nil
 		}}})
 
 	mux.Handle(ControlJobEndpointStatus,
 		// don't log requests to status endpoint, too spammy
-		jsonResponder{func() (interface{}, error) {
+		jsonResponder{log, func() (interface{}, error) {
 			s := j.jobs.status()
 			return s, nil
 		}})
 
 	mux.Handle(ControlJobEndpointSignal,
-		requestLogger{log: log, handler: jsonRequestResponder{func(decoder jsonDecoder) (interface{}, error) {
+		requestLogger{log: log, handler: jsonRequestResponder{log, func(decoder jsonDecoder) (interface{}, error) {
 			type reqT struct {
 				Name string
 				Op   string
@@ -136,8 +137,8 @@ func (j *controlJob) Run(ctx context.Context) {
 	server := http.Server{
 		Handler: mux,
 		// control socket is local, 1s timeout should be more than sufficient, even on a loaded system
-		WriteTimeout: 1*time.Second,
+		WriteTimeout: 1 * time.Second,
-		ReadTimeout: 1*time.Second,
+		ReadTimeout:  1 * time.Second,
 	}
 
 outer:
@@ -152,7 +153,10 @@ outer:
 		select {
 		case <-ctx.Done():
 			log.WithError(ctx.Err()).Info("context done")
-			server.Shutdown(context.Background())
+			err := server.Shutdown(context.Background())
+			if err != nil {
+				log.WithError(err).Error("cannot shutdown server")
+			}
 			break outer
 		case err = <-served:
 			if err != nil {
@@ -166,33 +170,50 @@ outer:
 }
 
 type jsonResponder struct {
+	log      Logger
 	producer func() (interface{}, error)
 }
 
 func (j jsonResponder) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	logIoErr := func(err error) {
+		if err != nil {
+			j.log.WithError(err).Error("control handler io error")
+		}
+	}
 	res, err := j.producer()
 	if err != nil {
+		j.log.WithError(err).Error("control handler error")
 		w.WriteHeader(http.StatusInternalServerError)
-		io.WriteString(w, err.Error())
+		_, err = io.WriteString(w, err.Error())
+		logIoErr(err)
 		return
 	}
 	var buf bytes.Buffer
 	err = json.NewEncoder(&buf).Encode(res)
 	if err != nil {
+		j.log.WithError(err).Error("control handler json marshal error")
 		w.WriteHeader(http.StatusInternalServerError)
-		io.WriteString(w, err.Error())
+		_, err = io.WriteString(w, err.Error())
 	} else {
-		io.Copy(w, &buf)
+		_, err = io.Copy(w, &buf)
 	}
+	logIoErr(err)
 }
 
 type jsonDecoder = func(interface{}) error
 
 type jsonRequestResponder struct {
+	log      Logger
 	producer func(decoder jsonDecoder) (interface{}, error)
 }
 
 func (j jsonRequestResponder) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	logIoErr := func(err error) {
+		if err != nil {
+			j.log.WithError(err).Error("control handler io error")
+		}
+	}
+
 	var decodeError error
 	decoder := func(i interface{}) error {
 		err := json.NewDecoder(r.Body).Decode(&i)
@@ -204,22 +225,28 @@ func (j jsonRequestResponder) ServeHTTP(w http.ResponseWriter, r *http.Request)
 	//If we had a decode error ignore output of producer and return error
 	if decodeError != nil {
 		w.WriteHeader(http.StatusBadRequest)
-		io.WriteString(w, decodeError.Error())
+		_, err := io.WriteString(w, decodeError.Error())
+		logIoErr(err)
 		return
 	}
 	if producerErr != nil {
+		j.log.WithError(producerErr).Error("control handler error")
 		w.WriteHeader(http.StatusInternalServerError)
-		io.WriteString(w, producerErr.Error())
+		_, err := io.WriteString(w, producerErr.Error())
+		logIoErr(err)
 		return
 	}
 
 	var buf bytes.Buffer
 	encodeErr := json.NewEncoder(&buf).Encode(res)
 	if encodeErr != nil {
+		j.log.WithError(producerErr).Error("control handler json marhsal error")
 		w.WriteHeader(http.StatusInternalServerError)
-		io.WriteString(w, encodeErr.Error())
+		_, err := io.WriteString(w, encodeErr.Error())
+		logIoErr(err)
 	} else {
-		io.Copy(w, &buf)
+		_, err := io.Copy(w, &buf)
+		logIoErr(err)
 	}
 }
 

daemon/daemon.go

@@ -3,8 +3,16 @@ package daemon
 import (
 	"context"
 	"fmt"
+	"os"
+	"os/signal"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
 	"github.com/zrepl/zrepl/daemon/job/reset"
@@ -12,12 +20,6 @@ import (
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/version"
-	"os"
-	"os/signal"
-	"strings"
-	"sync"
-	"syscall"
-	"time"
 )
 
 func Run(conf *config.Config) error {
@@ -74,12 +76,11 @@ func Run(conf *config.Config) error {
 			return errors.Errorf("unknown monitoring job #%d (type %T)", i, v)
 		}
 		if err != nil {
-			return errors.Wrapf(err,"cannot build monitorin gjob #%d", i)
+			return errors.Wrapf(err, "cannot build monitorin gjob #%d", i)
 		}
 		jobs.start(ctx, job, true)
 	}
 
-
 	log.Info("starting daemon")
 
 	// start regular jobs
@@ -117,8 +118,6 @@ func newJobs() *jobs {
 
 const (
 	logJobField string = "job"
-	logTaskField   string = "task"
-	logSubsysField string = "subsystem"
 )
 
 func (s *jobs) wait() <-chan struct{} {

daemon/filters/fsmapfilter.go

@@ -2,10 +2,12 @@ package filters
 
 import (
 	"fmt"
+	"strings"
+
 	"github.com/pkg/errors"
+
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/zfs"
-	"strings"
 )
 
 type DatasetMapFilter struct {

daemon/filters/fsvfilter.go

@@ -1,8 +1,9 @@
 package filters
 
 import (
-	"github.com/zrepl/zrepl/zfs"
 	"strings"
+
+	"github.com/zrepl/zrepl/zfs"
 )
 
 type AnyFSVFilter struct{}
@@ -17,7 +18,6 @@ func (AnyFSVFilter) Filter(t zfs.VersionType, name string) (accept bool, err err
 	return true, nil
 }
 
-
 type PrefixFilter struct {
 	prefix    string
 	fstype    zfs.VersionType

daemon/job/active.go

@@ -68,8 +68,7 @@ type activeSideTasks struct {
 func (a *ActiveSide) updateTasks(u func(*activeSideTasks)) activeSideTasks {
 	a.tasksMtx.Lock()
 	defer a.tasksMtx.Unlock()
-	var copy activeSideTasks
+	copy := a.tasks
-	copy = a.tasks
 	if u == nil {
 		return copy
 	}

daemon/job/build_jobs.go

@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	"github.com/pkg/errors"
+
 	"github.com/zrepl/zrepl/config"
 )
 
@@ -24,13 +25,13 @@ func JobsFromConfig(c *config.Config) ([]Job, error) {
 
 	// receiving-side root filesystems must not overlap
 	{
-		rfss := make([]string, len(js))
+		rfss := make([]string, 0, len(js))
-		for i, j := range js {
+		for _, j := range js {
 			jrfs, ok := j.OwnedDatasetSubtreeRoot()
 			if !ok {
 				continue
 			}
-			rfss[i] = jrfs.ToString()
+			rfss = append(rfss, jrfs.ToString())
 		}
 		if err := validateReceivingSidesDoNotOverlap(rfss); err != nil {
 			return nil, err

daemon/job/job.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/zfs"
 )
@@ -29,7 +30,6 @@ func WithLogger(ctx context.Context, l Logger) context.Context {
 	return context.WithValue(ctx, contextKeyLog, l)
 }
 
-
 type Job interface {
 	Name() string
 	Run(ctx context.Context)
@@ -65,7 +65,7 @@ func (s *Status) MarshalJSON() ([]byte, error) {
 	if err != nil {
 		return nil, err
 	}
-	m := map[string]json.RawMessage {
+	m := map[string]json.RawMessage{
 		"type":         typeJson,
 		string(s.Type): jobJSON,
 	}
@@ -94,16 +94,21 @@ func (s *Status) UnmarshalJSON(in []byte) (err error) {
 		var st SnapJobStatus
 		err = json.Unmarshal(jobJSON, &st)
 		s.JobSpecific = &st
-	case TypePull: fallthrough
+
+	case TypePull:
+		fallthrough
 	case TypePush:
 		var st ActiveSideStatus
 		err = json.Unmarshal(jobJSON, &st)
 		s.JobSpecific = &st
-	case TypeSource: fallthrough
+
+	case TypeSource:
+		fallthrough
 	case TypeSink:
 		var st PassiveStatus
 		err = json.Unmarshal(jobJSON, &st)
 		s.JobSpecific = &st
+
 	case TypeInternal:
 		// internal jobs do not report specifics
 	default:

daemon/job/snapjob.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/job/wakeup"

daemon/logging/logging_formatters.go

@@ -4,11 +4,13 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"time"
+
 	"github.com/fatih/color"
 	"github.com/go-logfmt/logfmt"
 	"github.com/pkg/errors"
+
 	"github.com/zrepl/zrepl/logger"
-	"time"
 )
 
 const (
@@ -159,10 +161,16 @@ func (f *LogfmtFormatter) Format(e *logger.Entry) ([]byte, error) {
 	enc := logfmt.NewEncoder(&buf)
 
 	if f.metadataFlags&MetadataTime != 0 {
-		enc.EncodeKeyval(FieldTime, e.Time)
+		err := enc.EncodeKeyval(FieldTime, e.Time)
+		if err != nil {
+			return nil, errors.Wrap(err, "logfmt: encode time")
+		}
 	}
 	if f.metadataFlags&MetadataLevel != 0 {
-		enc.EncodeKeyval(FieldLevel, e.Level)
+		err := enc.EncodeKeyval(FieldLevel, e.Level)
+		if err != nil {
+			return nil, errors.Wrap(err, "logfmt: encode level")
+		}
 	}
 
 	// at least try and put job and task in front
@@ -179,8 +187,10 @@ func (f *LogfmtFormatter) Format(e *logger.Entry) ([]byte, error) {
 		prefixed[pf] = true
 	}
 
-	enc.EncodeKeyval(FieldMessage, e.Message)
+	err := enc.EncodeKeyval(FieldMessage, e.Message)
-
+	if err != nil {
+		return nil, errors.Wrap(err, "logfmt: encode message")
+	}
 	for k, v := range e.Fields {
 		if !prefixed[k] {
 			if err := logfmtTryEncodeKeyval(enc, k, v); err != nil {
@@ -199,7 +209,10 @@ func logfmtTryEncodeKeyval(enc *logfmt.Encoder, field, value interface{}) error
 	case nil: // ok
 		return nil
 	case logfmt.ErrUnsupportedValueType:
-		enc.EncodeKeyval(field, fmt.Sprintf("<%T>", value))
+		err := enc.EncodeKeyval(field, fmt.Sprintf("<%T>", value))
+		if err != nil {
+			return errors.Wrap(err, "cannot encode unsuuported value type Go type")
+		}
 		return nil
 	}
 	return errors.Wrapf(err, "cannot encode field '%s'", field)

daemon/logging/logging_outlets.go

@@ -4,12 +4,14 @@ import (
 	"bytes"
 	"context"
 	"crypto/tls"
-	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/logger"
 	"io"
 	"log/syslog"
 	"net"
 	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/logger"
 )
 
 type EntryFormatter interface {
@@ -28,7 +30,10 @@ func (h WriterOutlet) WriteEntry(entry logger.Entry) error {
 		return err
 	}
 	_, err = h.writer.Write(bytes)
-	h.writer.Write([]byte("\n"))
+	if err != nil {
+		return err
+	}
+	_, err = h.writer.Write([]byte("\n"))
 	return err
 }
 
@@ -92,8 +97,10 @@ func (h *TCPOutlet) outLoop(retryInterval time.Duration) {
 				conn = nil
 			}
 		}
-		conn.SetWriteDeadline(time.Now().Add(retryInterval))
+		err = conn.SetWriteDeadline(time.Now().Add(retryInterval))
+		if err == nil {
 			_, err = io.Copy(conn, msg)
+		}
 		if err != nil {
 			retry = time.Now().Add(retryInterval)
 			conn.Close()

daemon/main.go

@@ -7,7 +7,7 @@ import (
 
 type Logger = logger.Logger
 
-var DaemonCmd = &cli.Subcommand {
+var DaemonCmd = &cli.Subcommand{
 	Use:   "daemon",
 	Short: "run the zrepl daemon",
 	Run: func(subcommand *cli.Subcommand, args []string) error {

daemon/nethelpers/helpers.go

@@ -1,10 +1,11 @@
 package nethelpers
 
 import (
-	"github.com/pkg/errors"
 	"net"
 	"os"
 	"path/filepath"
+
+	"github.com/pkg/errors"
 )
 
 func PreparePrivateSockpath(sockpath string) error {

daemon/pprof.go

@@ -7,11 +7,12 @@ import (
 	"context"
 	"net"
 	"net/http/pprof"
+
+	"github.com/zrepl/zrepl/daemon/job"
 )
 
 type pprofServer struct {
 	cc       chan PprofServerControlMsg
-	state    PprofServerControlMsg
 	listener net.Listener
 }
 
@@ -63,7 +64,14 @@ outer:
 			mux.Handle("/debug/pprof/profile", http.HandlerFunc(pprof.Profile))
 			mux.Handle("/debug/pprof/symbol", http.HandlerFunc(pprof.Symbol))
 			mux.Handle("/debug/pprof/trace", http.HandlerFunc(pprof.Trace))
-			go http.Serve(s.listener, mux)
+			go func() {
+				err := http.Serve(s.listener, mux)
+				if ctx.Err() != nil {
+					return
+				} else if err != nil {
+					job.GetLogger(ctx).WithError(err).Error("pprof server serve error")
+				}
+			}()
 			continue
 		}
 

daemon/prometheus.go

@@ -2,15 +2,17 @@ package daemon
 
 import (
 	"context"
+	"net"
+	"net/http"
+
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/rpc/dataconn/frameconn"
 	"github.com/zrepl/zrepl/zfs"
-	"net"
-	"net/http"
 )
 
 type prometheusJob struct {
@@ -63,17 +65,15 @@ func (j *prometheusJob) Run(ctx context.Context) {
 		log.WithError(err).Error("cannot listen")
 	}
 	go func() {
-		select {
+		<-ctx.Done()
-		case <-ctx.Done():
 		l.Close()
-		}
 	}()
 
 	mux := http.NewServeMux()
 	mux.Handle("/metrics", promhttp.Handler())
 
 	err = http.Serve(l, mux)
-	if err != nil {
+	if err != nil && ctx.Err() == nil {
 		log.WithError(err).Error("error while serving")
 	}
 
@@ -93,4 +93,3 @@ func (o prometheusJobOutlet) WriteEntry(entry logger.Entry) error {
 	prom.taskLogEntries.WithLabelValues(o.jobName, entry.Level.String()).Inc()
 	return nil
 }
-

daemon/pruner/pruner.go

@@ -3,17 +3,19 @@ package pruner
 import (
 	"context"
 	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/pruning"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/util/envconst"
-	"sort"
-	"strings"
-	"sync"
-	"time"
 )
 
 // Try to keep it compatible with gitub.com/zrepl/zrepl/endpoint.Endpoint
@@ -84,19 +86,6 @@ type LocalPrunerFactory struct {
 	promPruneSecs *prometheus.HistogramVec
 }
 
-func checkContainsKeep1(rules []pruning.KeepRule) error {
-	if len(rules) == 0 {
-		return nil //No keep rules means keep all - ok
-	}
-	for _, e := range rules {
-		switch e.(type) {
-		case *pruning.KeepLastN:
-			return nil
-		}
-	}
-	return errors.New("sender keep rules must contain last_n or be empty so that the last snapshot is definitely kept")
-}
-
 func NewLocalPrunerFactory(in config.PruningLocal, promPruneSecs *prometheus.HistogramVec) (*LocalPrunerFactory, error) {
 	rules, err := pruning.RulesFromConfig(in.Keep)
 	if err != nil {
@@ -139,7 +128,7 @@ func NewPrunerFactory(in config.PruningSenderReceiver, promPruneSecs *prometheus
 	f := &PrunerFactory{
 		senderRules:                    keepRulesSender,
 		receiverRules:                  keepRulesReceiver,
-		retryWait: envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10 * time.Second),
+		retryWait:                      envconst.Duration("ZREPL_PRUNER_RETRY_INTERVAL", 10*time.Second),
 		considerSnapAtCursorReplicated: considerSnapAtCursorReplicated,
 		promPruneSecs:                  promPruneSecs,
 	}
@@ -223,7 +212,7 @@ func (p *Pruner) prune(args args) {
 	// including figuring out how to resume a plan after being interrupted by network errors
 	// The non-retrying code in this package should move straight to replication/logic.
 	doOneAttempt(&args, u)
-	}
+}
 
 type Report struct {
 	State              string
@@ -326,7 +315,7 @@ func (f *fs) Report() FSReport {
 	}
 
 	r.DestroyList = make([]SnapshotReport, len(f.destroyList))
-	for i, snap := range f.destroyList{
+	for i, snap := range f.destroyList {
 		r.DestroyList[i] = snap.(snapshot).Report()
 	}
 
@@ -524,8 +513,7 @@ tfss_loop:
 		}
 	})
 
-
+}
-	}
 
 // attempts to exec pfs, puts it back into the queue with the result
 func doOneAttemptExec(a *args, u updater, pfs *fs) {
@@ -571,7 +559,7 @@ func doOneAttemptExec(a *args, u updater, pfs *fs) {
 		pairs := make([]string, len(destroyFails))
 		allSame := true
 		lastMsg := destroyFails[0].Error
-		for i := 0; i < len(destroyFails); i++{
+		for i := 0; i < len(destroyFails); i++ {
 			allSame = allSame && destroyFails[i].Error == lastMsg
 			relname := destroyFails[i].Snapshot.RelName()
 			names[i] = relname

daemon/pruner/pruner_queue.go

@@ -55,7 +55,7 @@ func (q *execQueue) Pop() *fs {
 	return fs
 }
 
-func(q *execQueue) Put(fs *fs, err error, done bool) {
+func (q *execQueue) Put(fs *fs, err error, done bool) {
 	fs.mtx.Lock()
 	fs.execErrLast = err
 	if done || err != nil {
@@ -79,5 +79,4 @@ func(q *execQueue) Put(fs *fs, err error, done bool) {
 	})
 	q.mtx.Unlock()
 
-
 }

daemon/snapper/snapper.go

@@ -1,18 +1,19 @@
 package snapper
 
 import (
-	"github.com/zrepl/zrepl/config"
-	"github.com/pkg/errors"
-	"time"
 	"context"
-	"github.com/zrepl/zrepl/daemon/filters"
 	"fmt"
-	"github.com/zrepl/zrepl/zfs"
 	"sort"
-	"github.com/zrepl/zrepl/logger"
 	"sync"
-)
+	"time"
 
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/logger"
+	"github.com/zrepl/zrepl/zfs"
+)
 
 //go:generate stringer -type=SnapState
 type SnapState uint
@@ -44,7 +45,7 @@ type args struct {
 	prefix         string
 	interval       time.Duration
 	fsf            *filters.DatasetMapFilter
-	snapshotsTaken chan<-struct{}
+	snapshotsTaken chan<- struct{}
 }
 
 type Snapper struct {
@@ -70,7 +71,7 @@ type Snapper struct {
 type State uint
 
 const (
-	SyncUp State = 1<<iota
+	SyncUp State = 1 << iota
 	SyncUpErrWait
 	Planning
 	Snapshotting
@@ -199,10 +200,10 @@ func syncUp(a args, u updater) state {
 	if err != nil {
 		return onErr(err, u)
 	}
-	u(func(s *Snapper){
+	u(func(s *Snapper) {
 		s.sleepUntil = syncPoint
 	})
-	t := time.NewTimer(syncPoint.Sub(time.Now()))
+	t := time.NewTimer(time.Until(syncPoint))
 	defer t.Stop()
 	select {
 	case <-t.C:
@@ -306,7 +307,7 @@ func wait(a args, u updater) state {
 		logFunc("enter wait-state after error")
 	})
 
-	t := time.NewTimer(sleepUntil.Sub(time.Now()))
+	t := time.NewTimer(time.Until(sleepUntil))
 	defer t.Stop()
 
 	select {
@@ -386,4 +387,3 @@ func findSyncPoint(log Logger, fss []*zfs.DatasetPath, prefix string, interval t
 	return snaptimes[0].time, nil
 
 }
-

daemon/snapper/snapper_all.go

@@ -3,6 +3,7 @@ package snapper
 import (
 	"context"
 	"fmt"
+
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
 )
@@ -17,7 +18,7 @@ type PeriodicOrManual struct {
 	s *Snapper
 }
 
-func (s *PeriodicOrManual) Run(ctx context.Context, wakeUpCommon chan <- struct{}) {
+func (s *PeriodicOrManual) Run(ctx context.Context, wakeUpCommon chan<- struct{}) {
 	if s.s != nil {
 		s.s.Run(ctx, wakeUpCommon)
 	}

docs/changelog.rst

@@ -25,6 +25,14 @@ We use the following annotations for classifying changes:
 * |bugfix| Change that fixes a bug, no regressions or incompatibilities expected.
 * |docs| Change to the documentation.
 
+0.1.1
+-----
+
+* |bugfix| :issue:`162` :commit:`d6304f4` : fix I/O timeout errors on variable receive rate
+
+  * A significant reduction or sudden stall of the receive rate (e.g. recv pool has other I/O to do)
+    would cause a ``writev I/O timeout`` error after approximately ten seconds.
+
 0.1
 ---
 
@@ -42,7 +50,7 @@ It breaks both configuration and transport format, and thus requires manual inte
 Notes to Package Maintainers
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 
-* Notify users about migrations (see changes attributed with |mig| below)
+* Notify users about config changes and migrations (see changes attributed with |break| and |mig| below)
 * If the daemon crashes, the stack trace produced by the Go runtime and possibly diagnostic output of zrepl will be written to stderr.
   This behavior is independent from the ``stdout`` outlet type.
   Please make sure the stderr output of the daemon is captured somewhere.
@@ -51,6 +59,7 @@ Notes to Package Maintainers
   This functionality will cause SIGABRT on panics and can be used to capture a coredump of the panicking process.
   To that extend, make sure that your package build system, your OS's coredump collection and the Go delve debugger work together.
   Use your build system to package the Go program in `this tutorial on Go coredumps and the delve debugger <https://rakyll.org/coredumps/>`_ , and make sure the symbol resolution etc. work on coredumps captured from the binary produced by your build system. (Special focus on symbol stripping, etc.)
+* Consider using the ``zrepl configcheck`` subcommand in startup scripts to abort a restart that would fail due to an invalid config.
 
 Changes
 ~~~~~~~

docs/index.rst

@@ -28,12 +28,32 @@ zrepl - ZFS replication
 
 .. raw:: html
 
-   <div style="margin-bottom: 1em; background: #2e3436; min-height: 6em; max-width: 100%">
+   <div style="margin-bottom: 1em; background: #2e3436; color: white; font-size: 0.8em; min-height: 6em; max-width: 100%; overflow: auto;">
-     <a href="https://raw.githubusercontent.com/wiki/zrepl/zrepl/zrepl_0.1_status.mp4" target="_new" >
+     <pre>
-       <video title="zrepl status subcommand" loop autoplay style="width: 100%; display: block;" src="https://raw.githubusercontent.com/wiki/zrepl/zrepl/zrepl_0.1_status.mp4"></video>
+     Job: prod_to_backups
-     </a>
+     Type: push
+     Replication:
+         Attempt #1
+         Status: fan-out-filesystems
+         Progress: [=========================\----] 246.7 MiB / 264.7 MiB @ 11.5 MiB/s
+           zroot              STEPPING (step 1/2, 624 B/1.2 KiB) next: @a => @b
+           zroot/ROOT         DONE (step 2/2, 1.2 KiB/1.2 KiB)
+           zroot/ROOT/default STEPPING (step 1/2, 123.4 MiB/129.3 MiB) next: @a => @b
+           zroot/tmp          STEPPING (step 1/2, 29.9 KiB/44.2 KiB) next: @a => @b
+           zroot/usr          STEPPING (step 1/2, 624 B/1.2 KiB) next: @a => @b
+           zroot/usr/home     STEPPING (step 1/2, 123.3 MiB/135.3 MiB) next: @a => @b
+           zroot/var          STEPPING (step 1/2, 624 B/1.2 KiB) next: @a => @b
+           zroot/var/audit    DONE (step 2/2, 1.2 KiB/1.2 KiB)
+           zroot/var/crash    DONE (step 2/2, 1.2 KiB/1.2 KiB)
+           zroot/var/log      STEPPING (step 1/2, 22.0 KiB/29.2 KiB) next: @a => @b
+           zroot/var/mail     STEPPING (step 1/2, 624 B/1.2 KiB) next: @a => @b
+     Pruning Sender:
+         ...
+     Pruning Receiver:
+     </pre>
    </div>
 
+
 Getting started
 ~~~~~~~~~~~~~~~
 

docs/publish.sh

@@ -54,9 +54,9 @@ echo "building site"
 set -e
 sphinx-versioning build \
    --show-banner \
-   --whitelist-branches 'master' \
+   --whitelist-branches '^master$' \
    --whitelist-tags '(v)?\d+\.\d+\.\d+$' \
-   --whitelist-tags '0.1.0-rc(3|4)$' \
+   --whitelist-tags '(v)?0.1.0-rc(3|4)$' \
    docs ./public_git \
    -- -c sphinxconf # older conf.py throw errors because they used
                     # version = subprocess.show_output(["git", "describe"])

docs/sphinxconf/conf.py

@@ -51,7 +51,7 @@ master_doc = 'index'
 
 # General information about the project.
 project = 'zrepl'
-copyright = '2017-2018, Christian Schwarz'
+copyright = '2017-2019, Christian Schwarz'
 author = 'Christian Schwarz'
 
 # The version info for the project you're documenting, acts as replacement for

endpoint/context.go

@@ -2,6 +2,7 @@ package endpoint
 
 import (
 	"context"
+
 	"github.com/zrepl/zrepl/logger"
 )
 

endpoint/endpoint.go

@@ -7,8 +7,11 @@ import (
 	"path"
 
 	"github.com/pkg/errors"
+
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/util/chainlock"
+	"github.com/zrepl/zrepl/util/envconst"
+	"github.com/zrepl/zrepl/util/semaphore"
 	"github.com/zrepl/zrepl/zfs"
 )
 
@@ -74,16 +77,29 @@ func (s *Sender) ListFilesystemVersions(ctx context.Context, r *pdu.ListFilesyst
 
 }
 
+var maxConcurrentZFSSendSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_SEND", 10))
+
 func (s *Sender) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.StreamCopier, error) {
 	_, err := s.filterCheckFS(r.Filesystem)
 	if err != nil {
 		return nil, nil, err
 	}
 
+	getLogger(ctx).Debug("acquire concurrent send semaphore")
+	// TODO use try-acquire and fail with resource-exhaustion rpc status
+	// => would require handling on the client-side
+	// => this is a dataconn endpoint, doesn't have the status code semantics of gRPC
+	guard, err := maxConcurrentZFSSendSemaphore.Acquire(ctx)
+	if err != nil {
+		return nil, nil, err
+	}
+	defer guard.Release()
+
 	si, err := zfs.ZFSSendDry(r.Filesystem, r.From, r.To, "")
 	if err != nil {
 		return nil, nil, err
 	}
+
 	var expSize int64 = 0      // protocol says 0 means no estimate
 	if si.SizeEstimate != -1 { // but si returns -1 for no size estimate
 		expSize = si.SizeEstimate
@@ -319,6 +335,8 @@ func (s *Receiver) Send(ctx context.Context, req *pdu.SendReq) (*pdu.SendRes, zf
 	return nil, nil, fmt.Errorf("receiver does not implement Send()")
 }
 
+var maxConcurrentZFSRecvSemaphore = semaphore.New(envconst.Int64("ZREPL_ENDPOINT_MAX_CONCURRENT_RECV", 10))
+
 func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs.StreamCopier) (*pdu.ReceiveRes, error) {
 	getLogger(ctx).Debug("incoming Receive")
 	defer receive.Close()
@@ -395,6 +413,16 @@ func (s *Receiver) Receive(ctx context.Context, req *pdu.ReceiveReq, receive zfs
 		}
 	}
 
+	getLogger(ctx).Debug("acquire concurrent recv semaphore")
+	// TODO use try-acquire and fail with resource-exhaustion rpc status
+	// => would require handling on the client-side
+	// => this is a dataconn endpoint, doesn't have the status code semantics of gRPC
+	guard, err := maxConcurrentZFSRecvSemaphore.Acquire(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer guard.Release()
+
 	getLogger(ctx).WithField("opts", fmt.Sprintf("%#v", recvOpts)).Debug("start receive command")
 
 	if err := zfs.ZFSRecv(ctx, lp.ToString(), receive, recvOpts); err != nil {

lazy.sh

@@ -39,7 +39,9 @@ godep() {
     go build -o "$GOPATH/bin/stringer"      ./vendor/golang.org/x/tools/cmd/stringer
     go build -o "$GOPATH/bin/protoc-gen-go" ./vendor/github.com/golang/protobuf/protoc-gen-go
     go build -o "$GOPATH/bin/enumer"        ./vendor/github.com/alvaroloes/enumer
-    if ! type stringer || ! type protoc-gen-go || ! type enumer ; then
+    go build -o "$GOPATH/bin/goimports"     ./vendor/golang.org/x/tools/cmd/goimports
+    go build -o "$GOPATH/bin/golangci-lint" ./vendor/github.com/golangci/golangci-lint/cmd/golangci-lint
+    if ! type stringer || ! type protoc-gen-go || ! type enumer || ! type goimports || ! type golangci-lint; then
         echo "Installed dependencies but can't find them in \$PATH, adjust it to contain \$GOPATH/bin" 1>&2
         exit 1
     fi

logger/datastructures.go

@@ -4,10 +4,11 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
-	"github.com/fatih/color"
-	"github.com/pkg/errors"
 	"sync"
 	"time"
+
+	"github.com/fatih/color"
+	"github.com/pkg/errors"
 )
 
 type Level int
@@ -66,7 +67,7 @@ func (l Level) Short() string {
 	case Error:
 		return "ERRO"
 	default:
-		return fmt.Sprintf("%s", l)
+		return l.String()
 	}
 }
 
@@ -81,7 +82,7 @@ func (l Level) String() string {
 	case Error:
 		return "error"
 	default:
-		return fmt.Sprintf("%s", string(l))
+		return string(l)
 	}
 }
 

logger/logger.go

@@ -66,7 +66,8 @@ func (l *loggerImpl) logInternalError(outlet Outlet, err string) {
 		time.Now(),
 		fields,
 	}
-	l.outlets.GetLoggerErrorOutlet().WriteEntry(entry)
+	// ignore errors at this point (still better than panicking if the error is temporary)
+	_ = l.outlets.GetLoggerErrorOutlet().WriteEntry(entry)
 }
 
 func (l *loggerImpl) log(level Level, msg string) {

logger/logger_test.go

@@ -2,10 +2,12 @@ package logger_test
 
 import (
 	"fmt"
-	"github.com/kr/pretty"
-	"github.com/zrepl/zrepl/logger"
 	"testing"
 	"time"
+
+	"github.com/kr/pretty"
+
+	"github.com/zrepl/zrepl/logger"
 )
 
 type TestOutlet struct {

logger/stderrlogger.go

@@ -5,11 +5,7 @@ import (
 	"os"
 )
 
-type stderrLogger struct {
+type stderrLoggerOutlet struct{}
-	Logger
-}
-
-type stderrLoggerOutlet struct {}
 
 func (stderrLoggerOutlet) WriteEntry(entry Entry) error {
 	fmt.Fprintf(os.Stderr, "%#v\n", entry)

pruning/keep_grid.go

@@ -2,12 +2,14 @@ package pruning
 
 import (
 	"fmt"
-	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/pruning/retentiongrid"
 	"regexp"
 	"sort"
 	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/pruning/retentiongrid"
 )
 
 // KeepGrid fits snapshots that match a given regex into a retentiongrid.Grid,

pruning/keep_helpers_test.go

@@ -1,8 +1,9 @@
 package pruning
 
 import (
-	"github.com/stretchr/testify/assert"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestShallowCopySnapList(t *testing.T) {

pruning/keep_last_n.go

@@ -1,8 +1,9 @@
 package pruning
 
 import (
-	"github.com/pkg/errors"
 	"sort"
+
+	"github.com/pkg/errors"
 )
 
 type KeepLastN struct {

pruning/keep_last_n_test.go

@@ -1,9 +1,10 @@
 package pruning
 
 import (
-	"github.com/stretchr/testify/assert"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestKeepLastN(t *testing.T) {

pruning/keep_not_replicated.go

@@ -1,8 +1,6 @@
 package pruning
 
-type KeepNotReplicated struct {
+type KeepNotReplicated struct{}
-	forceConstructor struct{}
-}
 
 func (*KeepNotReplicated) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
 	return filterSnapList(snaps, func(snapshot Snapshot) bool {

pruning/pruning.go

@@ -2,9 +2,11 @@ package pruning
 
 import (
 	"fmt"
-	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/config"
 	"time"
+
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/config"
 )
 
 type KeepRule interface {
@@ -20,7 +22,7 @@ type Snapshot interface {
 // The returned snapshot list is guaranteed to only contains elements of input parameter snaps
 func PruneSnapshots(snaps []Snapshot, keepRules []KeepRule) []Snapshot {
 
-	if keepRules == nil || len(keepRules) == 0 {
+	if len(keepRules) == 0 {
 		return []Snapshot{}
 	}
 

pruning/retentiongrid/retentiongrid_test.go

@@ -2,11 +2,12 @@ package retentiongrid
 
 import (
 	"fmt"
-	"github.com/stretchr/testify/assert"
 	"strconv"
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 type retentionIntervalStub struct {

replication/driver/errorclass_enumer.go

@@ -6,9 +6,9 @@ import (
 	"fmt"
 )
 
-const _errorClassName = "errorClassUnknownerrorClassPermanenterrorClassTemporaryConnectivityRelated"
+const _errorClassName = "errorClassPermanenterrorClassTemporaryConnectivityRelated"
 
-var _errorClassIndex = [...]uint8{0, 17, 36, 74}
+var _errorClassIndex = [...]uint8{0, 19, 57}
 
 func (i errorClass) String() string {
 	if i < 0 || i >= errorClass(len(_errorClassIndex)-1) {
@@ -17,12 +17,11 @@ func (i errorClass) String() string {
 	return _errorClassName[_errorClassIndex[i]:_errorClassIndex[i+1]]
 }
 
-var _errorClassValues = []errorClass{0, 1, 2}
+var _errorClassValues = []errorClass{0, 1}
 
 var _errorClassNameToValueMap = map[string]errorClass{
-	_errorClassName[0:17]:  0,
+	_errorClassName[0:19]:  0,
-	_errorClassName[17:36]: 1,
+	_errorClassName[19:57]: 1,
-	_errorClassName[36:74]: 2,
 }
 
 // errorClassString retrieves an enum value from the enum constants string name.

replication/driver/replication_driver.go

@@ -10,11 +10,12 @@ import (
 	"sync"
 	"time"
 
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
 	"github.com/zrepl/zrepl/replication/report"
 	"github.com/zrepl/zrepl/util/chainlock"
 	"github.com/zrepl/zrepl/util/envconst"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
 )
 
 type interval struct {
@@ -338,7 +339,7 @@ func (a *attempt) do(ctx context.Context, prev *attempt) {
 				l := fmt.Sprintf("  %s => %v", i.cur.fs.ReportInfo().Name, prevNames)
 				inconsistencyLines = append(inconsistencyLines, l)
 			}
-			fmt.Fprintf(&msg, strings.Join(inconsistencyLines, "\n"))
+			fmt.Fprint(&msg, strings.Join(inconsistencyLines, "\n"))
 			now := time.Now()
 			a.planErr = newTimedError(errors.New(msg.String()), now)
 			a.fss = nil
@@ -551,17 +552,11 @@ func (s *step) report() *report.StepReport {
 	return r
 }
 
-type stepErrorReport struct {
-	err  *timedError
-	step int
-}
-
 //go:generate enumer -type=errorClass
 type errorClass int
 
 const (
-	errorClassUnknown errorClass = iota
+	errorClassPermanent errorClass = iota
-	errorClassPermanent
 	errorClassTemporaryConnectivityRelated
 )
 

replication/driver/replication_driver_debug.go

@@ -13,6 +13,7 @@ func init() {
 	}
 }
 
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "repl: driver: %s\n", fmt.Sprintf(format, args...))
@@ -21,6 +22,7 @@ func debug(format string, args ...interface{}) {
 
 type debugFunc func(format string, args ...interface{})
 
+//nolint[:deadcode,unused]
 func debugPrefix(prefixFormat string, prefixFormatArgs ...interface{}) debugFunc {
 	prefix := fmt.Sprintf(prefixFormat, prefixFormatArgs...)
 	return func(format string, args ...interface{}) {

replication/driver/replication_driver_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+
 	"github.com/zrepl/zrepl/replication/report"
 
 	"github.com/stretchr/testify/assert"
@@ -165,14 +166,14 @@ func TestReplication(t *testing.T) {
 	reports := make([]*report.Report, len(fireAt))
 	for i := range fireAt {
 		sleepUntil := begin.Add(fireAt[i])
-		time.Sleep(sleepUntil.Sub(time.Now()))
+		time.Sleep(time.Until(sleepUntil))
 		reports[i] = getReport()
 		// uncomment for viewing non-diffed results
 		// t.Logf("report @ %6.4f:\n%s", fireAt[i].Seconds(), pretty.Sprint(reports[i]))
 	}
 	waitBegin := time.Now()
 	wait(true)
-	waitDuration := time.Now().Sub(waitBegin)
+	waitDuration := time.Since(waitBegin)
 	assert.True(t, waitDuration < 10*time.Millisecond, "%v", waitDuration) // and that's gratious
 
 	prev, err := json.Marshal(reports[0])

replication/driver/replication_stepqueue_test.go

@@ -96,7 +96,7 @@ func TestPqConcurrent(t *testing.T) {
 				pos := atomic.AddUint32(&globalCtr, 1)
 				t := time.Unix(int64(step), 0)
 				done := q.WaitReady(fs, t)
-				wakeAt := time.Now().Sub(begin)
+				wakeAt := time.Since(begin)
 				time.Sleep(sleepTimePerStep)
 				done()
 				recs = append(recs, record{fs, step, pos, wakeAt})

replication/logic/diff/diff_test.go

@@ -97,7 +97,7 @@ func TestIncrementalPath_SnapshotsOnly(t *testing.T) {
 	})
 
 	// sender with earlier but also current version as sender is not a conflict
-	doTest(l("@c,3"), l("@a,1", "@b,2", "@c,3") , func(path []*FilesystemVersion, conflict error) {
+	doTest(l("@c,3"), l("@a,1", "@b,2", "@c,3"), func(path []*FilesystemVersion, conflict error) {
 		t.Logf("path: %#v", path)
 		t.Logf("conflict: %#v", conflict)
 		assert.Empty(t, path)

replication/logic/pdu/pdu_extras.go

@@ -2,8 +2,9 @@ package pdu
 
 import (
 	"fmt"
-	"github.com/zrepl/zrepl/zfs"
 	"time"
+
+	"github.com/zrepl/zrepl/zfs"
 )
 
 func (v *FilesystemVersion) RelName() string {

replication/logic/pdu/pdu_test.go

@@ -1,9 +1,10 @@
 package pdu
 
 import (
-	"github.com/stretchr/testify/assert"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestFilesystemVersion_RelName(t *testing.T) {
@@ -58,7 +59,7 @@ func TestFilesystemVersion_RelName(t *testing.T) {
 func TestFilesystemVersion_ZFSFilesystemVersion(t *testing.T) {
 
 	empty := &FilesystemVersion{}
-	_, err:= empty.ZFSFilesystemVersion()
+	_, err := empty.ZFSFilesystemVersion()
 	assert.Error(t, err)
 
 	dateInvalid := &FilesystemVersion{Creation: "foobar"}

replication/logic/replication_logic.go

@@ -14,6 +14,8 @@ import (
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/replication/report"
 	"github.com/zrepl/zrepl/util/bytecounter"
+	"github.com/zrepl/zrepl/util/envconst"
+	"github.com/zrepl/zrepl/util/semaphore"
 	"github.com/zrepl/zrepl/zfs"
 )
 
@@ -26,7 +28,7 @@ type Endpoint interface {
 	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
 	ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
 	DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error)
-	WaitForConnectivity(ctx context.Context) (error)
+	WaitForConnectivity(ctx context.Context) error
 }
 
 type Sender interface {
@@ -110,6 +112,8 @@ type Filesystem struct {
 	Path                string // compat
 	receiverFS          *pdu.Filesystem
 	promBytesReplicated prometheus.Counter // compat
+
+	sizeEstimateRequestSem *semaphore.S
 }
 
 func (f *Filesystem) EqualToPreviousAttempt(other driver.FS) bool {
@@ -234,6 +238,8 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
 	}
 	rfss := rlfssres.GetFilesystems()
 
+	sizeEstimateRequestSem := semaphore.New(envconst.Int64("ZREPL_REPLICATION_MAX_CONCURRENT_SIZE_ESTIMATE", 4))
+
 	q := make([]*Filesystem, 0, len(sfss))
 	for _, fs := range sfss {
 
@@ -252,6 +258,7 @@ func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
 			Path:                   fs.Path,
 			receiverFS:             receiverFS,
 			promBytesReplicated:    ctr,
+			sizeEstimateRequestSem: sizeEstimateRequestSem,
 		})
 	}
 
@@ -336,7 +343,17 @@ func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
 		wg.Add(1)
 		go func(step *Step) {
 			defer wg.Done()
-			err := step.updateSizeEstimate(fanOutCtx)
+
+			// TODO instead of the semaphore, rely on resource-exhaustion signaled by the remote endpoint to limit size-estimate requests
+			// Send is handled over rpc/dataconn ATM, which doesn't support the resource exhaustion status codes that gRPC defines
+			guard, err := fs.sizeEstimateRequestSem.Acquire(fanOutCtx)
+			if err != nil {
+				fanOutCancel()
+				return
+			}
+			defer guard.Release()
+
+			err = step.updateSizeEstimate(fanOutCtx)
 			if err != nil {
 				log.WithError(err).WithField("step", step).Error("error computing size estimate")
 				fanOutCancel()

rpc/dataconn/dataconn_client.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/golang/protobuf/proto"
+
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn/stream"
 	"github.com/zrepl/zrepl/transport"
@@ -214,7 +215,6 @@ func (c *Client) ReqRecv(ctx context.Context, req *pdu.ReceiveReq, streamCopier
 	return res.res, cause
 }
 
-
 func (c *Client) ReqPing(ctx context.Context, req *pdu.PingReq) (*pdu.PingRes, error) {
 	conn, err := c.getWire(ctx)
 	if err != nil {

rpc/dataconn/dataconn_debug.go

@@ -13,6 +13,7 @@ func init() {
 	}
 }
 
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "rpc/dataconn: %s\n", fmt.Sprintf(format, args...))

rpc/dataconn/dataconn_server.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"github.com/golang/protobuf/proto"
+
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn/stream"
@@ -137,7 +138,6 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 	default:
 		s.log.WithField("endpoint", endpoint).Error("unknown endpoint")
 		handlerErr = fmt.Errorf("requested endpoint does not exist")
-		return
 	}
 
 	s.log.WithField("endpoint", endpoint).WithField("errType", fmt.Sprintf("%T", handlerErr)).Debug("handler returned")
@@ -187,6 +187,4 @@ func (s *Server) serveConn(nc *transport.AuthConn) {
 			s.log.WithError(err).Error("cannot write send stream")
 		}
 	}
-
-	return
 }

rpc/dataconn/frameconn/frameconn.go

@@ -1,7 +1,6 @@
 package frameconn
 
 import (
-	"bufio"
 	"encoding/binary"
 	"errors"
 	"fmt"
@@ -12,6 +11,7 @@ import (
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/zrepl/zrepl/rpc/dataconn/base2bufpool"
 	"github.com/zrepl/zrepl/rpc/dataconn/timeoutconn"
 )
@@ -47,7 +47,6 @@ func (f *FrameHeader) Unmarshal(buf []byte) {
 type Conn struct {
 	readMtx, writeMtx sync.Mutex
 	nc                timeoutconn.Conn
-	ncBuf             *bufio.ReadWriter
 	readNextValid     bool
 	readNext          FrameHeader
 	nextReadErr       error
@@ -223,6 +222,10 @@ func (c *Conn) writeFrame(payload []byte, frameType uint32) error {
 	return nil
 }
 
+func (c *Conn) ResetWriteTimeout() error {
+	return c.nc.RenewWriteDeadline()
+}
+
 func (c *Conn) Shutdown(deadline time.Time) error {
 	// TCP connection teardown is a bit wonky if we are in a situation
 	// where there is still data in flight (DIF) to our side:

rpc/dataconn/frameconn/frameconn_shutdown_fsm.go

@@ -10,17 +10,11 @@ type shutdownFSM struct {
 type shutdownFSMState uint32
 
 const (
+	// zero value is important
 	shutdownStateOpen shutdownFSMState = iota
 	shutdownStateBegin
 )
 
-func newShutdownFSM() *shutdownFSM {
-	fsm := &shutdownFSM{
-		state:          shutdownStateOpen,
-	}
-	return fsm
-}
-
 func (f *shutdownFSM) Begin() (thisCallStartedShutdown bool) {
 	f.mtx.Lock()
 	defer f.mtx.Unlock()
@@ -34,4 +28,3 @@ func (f *shutdownFSM) IsShuttingDown() bool {
 	defer f.mtx.Unlock()
 	return f.state != shutdownStateOpen
 }
-

rpc/dataconn/frameconn/frameconn_test.go

@@ -19,4 +19,3 @@ func TestIsPublicFrameType(t *testing.T) {
 	assert.True(t, IsPublicFrameType(255))
 	assert.False(t, IsPublicFrameType(rstFrameType))
 }
-

rpc/dataconn/heartbeatconn/heartbeatconn.go

@@ -12,8 +12,6 @@ import (
 
 type Conn struct {
 	state                 state
-	// if not nil, opErr is returned for ReadFrame and WriteFrame (not for Close, though)
-	opErr                 atomic.Value // error
 	fc                    *frameconn.Conn
 	sendInterval, timeout time.Duration
 	stopSend              chan struct{}
@@ -97,7 +95,10 @@ func (c *Conn) sendHeartbeats() {
 				debug("send heartbeat")
 				// if the connection is in zombie mode (aka iptables DROP inbetween peers)
 				// this call or one of its successors will block after filling up the kernel tx buffer
-				c.fc.WriteFrame([]byte{}, heartbeat)
+				err := c.fc.WriteFrame([]byte{}, heartbeat)
+				if err != nil {
+					debug("send heartbeat error: %s", err)
+				}
 				// ignore errors from WriteFrame to rate-limit SendHeartbeat retries
 				c.lastFrameSent.Store(time.Now())
 			}()
@@ -122,7 +123,17 @@ func (c *Conn) readFrameFiltered() (frameconn.Frame, error) {
 			return frameconn.Frame{}, fmt.Errorf("unknown frame type %x", f.Header.Type)
 		}
 		// drop heartbeat frame
-		debug("received heartbeat")
+		debug("received heartbeat, resetting write timeout")
+		// the peer's heartbeat proves to us that the peer is still live
+		// => trust the peer at this point (DoS risks are ignored ATM)
+		// => we assume that the connection is symmetric duplex, i.e., if receiving works for us,
+		//    sending works for us, too.
+		// So, let's grant the peer another write timeout.
+		err = c.fc.ResetWriteTimeout()
+		debug("renew frameconn write timeout returned errT=%T err=%s", err, err)
+		if err != nil {
+			return frameconn.Frame{}, err
+		}
 		continue
 	}
 }

rpc/dataconn/heartbeatconn/heartbeatconn_debug.go

@@ -13,6 +13,7 @@ func init() {
 	}
 }
 
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "rpc/dataconn/heartbeatconn: %s\n", fmt.Sprintf(format, args...))

rpc/dataconn/heartbeatconn/heartbeatconn_test.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+
 	"github.com/zrepl/zrepl/rpc/dataconn/frameconn"
 )
 

rpc/dataconn/heartbeatconn/integration_test_variablereceiverate/heartbeatconn_integration_variablereceiverate.go

@@ -0,0 +1,144 @@
+// This integration test exercises the behavior of heartbeatconn
+// where the server is slow at handling the data received over the connection.
+// Note that the server is still sending heartbeats to the client, it's just the
+// data handling (usually I/O in case of zrepl endpoint.Receiver) that is slow.
+//
+// In commit 082335df5d85e1b0b9faa35ff182c71886142d3e and earlier, heartbeatconn would fail
+// this benchmark with a writev I/O timeout (here the ss(8) output at the time of failure)
+//
+//  ESTAB        33369    0                                        127.0.0.1:12345                                                127.0.0.1:57282                    users:(("heartbeatconn_i",pid=25953,fd=5))
+//	 cubic wscale:7,7 rto:203 rtt:2.992/5.849 ato:162 mss:32768 pmtu:65535 rcvmss:32741 advmss:65483 cwnd:10 bytes_sent:48 bytes_acked:48 bytes_received:195401 segs_out:44 segs_in:57 data_segs_out:6 data_segs_in:34 send 876.1Mbps lastsnd:125 lastrcv:9390 lastack:125 pacing_rate 1752.0Mbps delivery_rate 6393.8Mbps delivered:7 app_limited busy:42ms rcv_rtt:1 rcv_space:65483 rcv_ssthresh:65483 minrtt:0.029
+//	 --
+//	 ESTAB        0        3956805                                  127.0.0.1:57282                                                127.0.0.1:12345                    users:(("heartbeatconn_i",pid=26100,fd=3))
+//		  cubic wscale:7,7 rto:211 backoff:5 rtt:10.38/16.937 ato:40 mss:32768 pmtu:65535 rcvmss:536 advmss:65483 cwnd:10 bytes_sent:195401 bytes_acked:195402 bytes_received:48 segs_out:57 segs_in:45 data_segs_out:34 data_segs_in:6 send 252.5Mbps lastsnd:9390 lastrcv:125 lastack:125 pacing_rate 505.1Mbps delivery_rate 1971.0Mbps delivered:35 busy:30127ms rwnd_limited:30086ms(99.9%) rcv_space:65495 rcv_ssthresh:65495 notsent:3956805 minrtt:0.007
+//	 panic: writev tcp 127.0.0.1:57282->127.0.0.1:12345: i/o timeout
+//
+// The assumed reason for those writev timeouts is the following:
+// - Sporadic server stalls (sever data handling, usually I/O) cause TCP exponential backoff on the client for client->server
+// - Go runtime unblocks after the deadline expires, resultin gin writev I/O timeout
+// - That is, even though the client observed heartbeats from the server
+// -> TCP doesn't assume symmetric connection behavior, but our implementation does.
+//
+// The fix contained in the commit this message was committed with resets the deadline whenever
+// a heartbeat is received from the server.
+//
+//
+// How to run this integration test:
+//
+//
+//  Terminal 1:
+//     $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go  -mode server -addr 127.0.0.1:12345
+//     rpc/dataconn/heartbeatconn: send heartbeat
+//     rpc/dataconn/heartbeatconn: send heartbeat
+//     ...
+//
+//  Terminal 2:
+//     $ ZREPL_RPC_DATACONN_HEARTBEATCONN_DEBUG=1 go run heartbeatconn_integration_variablereceiverate.go  -mode  client -addr 127.0.0.1:12345
+//     rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
+//     rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
+//     rpc/dataconn/heartbeatconn: send heartbeat
+//     rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
+//     rpc/dataconn/heartbeatconn: renew frameconn write timeout returned errT=<nil> err=%!s(<nil>)
+//     rpc/dataconn/heartbeatconn: received heartbeat, resetting write timeout
+//     ...
+//
+//  You should observe
+package main
+
+import (
+	"bytes"
+	"flag"
+	"fmt"
+	"io"
+	"log"
+	"math/rand"
+	"net"
+	"os"
+	"os/exec"
+	"path"
+	"time"
+
+	"github.com/zrepl/zrepl/util/devnoop"
+
+	"github.com/zrepl/zrepl/rpc/dataconn/heartbeatconn"
+)
+
+func orDie(err error) {
+	if err != nil {
+		grepfield := path.Base(os.Args[0])[:10]
+		fmt.Fprintf(os.Stderr, "grepping for %s\n", grepfield)
+		sh := fmt.Sprintf("ss -ntpi | grep -A1 %s", grepfield)
+		cmd := exec.Command("bash", "-c", sh)
+		o, _ := cmd.CombinedOutput()
+		buf := bytes.NewBuffer(o)
+		_, _ = io.Copy(os.Stderr, buf)
+		panic(err)
+	}
+}
+
+var mode string
+var addr string
+
+func main() {
+
+	flag.StringVar(&mode, "mode", "", "server|client")
+	flag.StringVar(&addr, "addr", "INVALID", "")
+	flag.Parse()
+
+	modemap := map[string]func(){
+		"server": server,
+		"client": client,
+	}
+	modemap[mode]()
+
+}
+
+func server() {
+	ln, err := net.Listen("tcp", addr)
+	orDie(err)
+	l := ln.(*net.TCPListener)
+	for {
+		c, err := l.AcceptTCP()
+		if err != nil {
+			log.Printf("accept err: %s", err)
+			continue
+		}
+		hc := heartbeatconn.Wrap(c, 5*time.Second, 10*time.Second)
+
+		for {
+			f, err := hc.ReadFrame()
+			orDie(err)
+			// _, err = buf.Write(f.Buffer.Bytes())
+			// orDie(err)
+			sleep := time.Duration(rand.NormFloat64()*500) * time.Millisecond
+			time.Sleep(sleep)
+			f.Buffer.Free()
+		}
+
+	}
+
+}
+
+func client() {
+	c, err := net.Dial("tcp", addr)
+	orDie(err)
+	hc := heartbeatconn.Wrap(c.(*net.TCPConn), 5*time.Second, 10*time.Second)
+
+	// follow API requirements to always ReadFrame
+	go func() {
+		for {
+			f, err := hc.ReadFrame()
+			orDie(err)
+			f.Buffer.Free()
+		}
+	}()
+
+	dn := devnoop.Get()
+	var buf [1 << 10]byte
+	for {
+		n, err := dn.Read(buf[:])
+		orDie(err)
+		err = hc.WriteFrame(buf[:n], 23)
+		orDie(err)
+	}
+}

rpc/dataconn/stream/stream.go

@@ -28,6 +28,7 @@ func WithLogger(ctx context.Context, log Logger) context.Context {
 	return context.WithValue(ctx, contextKeyLogger, log)
 }
 
+//nolint[:deadcode,unused]
 func getLog(ctx context.Context) Logger {
 	log, ok := ctx.Value(contextKeyLogger).(Logger)
 	if !ok {

rpc/dataconn/stream/stream_conn.go

@@ -25,7 +25,6 @@ type Conn struct {
 	// support a single stream at a time over hc.
 	readMtx   sync.Mutex
 	readClean bool
-	allowWriteStreamTo bool
 
 	// writeMtx serializes write stream operations because we inherently only
 	// support a single stream at a time over hc.
@@ -95,7 +94,7 @@ func (c *Conn) ReadStreamedMessage(ctx context.Context, maxSize uint32, frameTyp
 	}()
 	err := readStream(c.frameReads, c.hc, w, frameType)
 	c.readClean = isConnCleanAfterRead(err)
-	w.CloseWithError(readMessageSentinel)
+	_ = w.CloseWithError(readMessageSentinel) // always returns nil
 	wg.Wait()
 	if err != nil {
 		return nil, err
@@ -166,7 +165,7 @@ func (c *Conn) SendStream(ctx context.Context, src zfs.StreamCopier, frameType u
 		var res writeStreamRes
 		res.errStream, res.errConn = writeStream(ctx, c.hc, r, frameType)
 		if w != nil {
-			w.CloseWithError(res.errStream)
+			_ = w.CloseWithError(res.errStream) // always returns nil
 		}
 		writeStreamErrChan <- res
 	}()

rpc/dataconn/stream/stream_debug.go

@@ -13,6 +13,7 @@ func init() {
 	}
 }
 
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "rpc/dataconn/stream: %s\n", fmt.Sprintf(format, args...))

rpc/dataconn/stream/stream_test.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/rpc/dataconn/heartbeatconn"
 	"github.com/zrepl/zrepl/util/socketpair"

rpc/dataconn/timeoutconn/internal/wireevaluator/wireevaluator.go

@@ -11,6 +11,7 @@ import (
 
 	netssh "github.com/problame/go-netssh"
 	"github.com/zrepl/yaml-config"
+
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/transport"
 	transportconfig "github.com/zrepl/zrepl/transport/fromconfig"

rpc/dataconn/timeoutconn/internal/wireevaluator/wireevaluator_closewrite.go

@@ -52,9 +52,6 @@ func (CloseWrite) sender(wire transport.Wire) {
 		log.Printf("closeErr=%T %s", closeErr, closeErr)
 	}()
 
-	type opResult struct {
-		err error
-	}
 	writeDone := make(chan struct{}, 1)
 	go func() {
 		close(writeDone)
@@ -85,7 +82,7 @@ func (CloseWrite) receiver(wire transport.Wire) {
 
 	// consume half the test data, then detect an error, send it and CloseWrite
 
-	r := io.LimitReader(wire, int64(5 * len(closeWriteTestSendData)/3))
+	r := io.LimitReader(wire, int64(5*len(closeWriteTestSendData)/3))
 	_, err := io.Copy(ioutil.Discard, r)
 	noerror(err)
 

rpc/dataconn/timeoutconn/timeoutconn.go

@@ -80,7 +80,7 @@ func (c *Conn) renewReadDeadline() error {
 	return c.SetReadDeadline(time.Now().Add(c.idleTimeout))
 }
 
-func (c *Conn) renewWriteDeadline() error {
+func (c *Conn) RenewWriteDeadline() error {
 	if atomic.LoadInt32(&c.renewDeadlinesDisabled) != 0 {
 		return nil
 	}
@@ -95,7 +95,7 @@ restart:
 		return n, err
 	}
 	var nCurRead int
-	nCurRead, err = c.Wire.Read(p[n:len(p)])
+	nCurRead, err = c.Wire.Read(p[n:])
 	n += nCurRead
 	if netErr, ok := err.(net.Error); ok && netErr.Timeout() && nCurRead > 0 {
 		err = nil
@@ -107,11 +107,11 @@ restart:
 func (c Conn) Write(p []byte) (n int, err error) {
 	n = 0
 restart:
-	if err := c.renewWriteDeadline(); err != nil {
+	if err := c.RenewWriteDeadline(); err != nil {
 		return n, err
 	}
 	var nCurWrite int
-	nCurWrite, err = c.Wire.Write(p[n:len(p)])
+	nCurWrite, err = c.Wire.Write(p[n:])
 	n += nCurWrite
 	if netErr, ok := err.(net.Error); ok && netErr.Timeout() && nCurWrite > 0 {
 		err = nil
@@ -127,7 +127,7 @@ restart:
 func (c Conn) WritevFull(bufs net.Buffers) (n int64, err error) {
 	n = 0
 restart:
-	if err := c.renewWriteDeadline(); err != nil {
+	if err := c.RenewWriteDeadline(); err != nil {
 		return n, err
 	}
 	var nCurWrite int64

rpc/dataconn/timeoutconn/timeoutconn_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
 	"github.com/zrepl/zrepl/util/socketpair"
 )
 
@@ -101,7 +102,7 @@ func TestNoPartialReadsDueToDeadline(t *testing.T) {
 		// io.Copy will encounter a partial read, then wait ~50ms until the other 5 bytes are written
 		// It is still going to fail with deadline err because it expects EOF
 		n, err := io.Copy(&buf, bc)
-		readDuration := time.Now().Sub(beginRead)
+		readDuration := time.Since(beginRead)
 		t.Logf("read duration=%s", readDuration)
 		t.Logf("recv done n=%v err=%v", n, err)
 		t.Logf("buf=%v", buf.Bytes())
@@ -152,7 +153,7 @@ func TestPartialWriteMockConn(t *testing.T) {
 	buf := []byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
 	begin := time.Now()
 	n, err := mc.Write(buf[:])
-	duration := time.Now().Sub(begin)
+	duration := time.Since(begin)
 	assert.NoError(t, err)
 	assert.Equal(t, 5, n)
 	assert.True(t, duration > 100*time.Millisecond)

rpc/grpcclientidentity/authlistener_grpc_adaptor.go

@@ -18,11 +18,12 @@ import (
 	"net"
 	"time"
 
-	"github.com/zrepl/zrepl/logger"
-	"github.com/zrepl/zrepl/transport"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/peer"
+
+	"github.com/zrepl/zrepl/logger"
+	"github.com/zrepl/zrepl/transport"
 )
 
 type Logger = logger.Logger
@@ -105,7 +106,7 @@ func NewInterceptors(logger Logger, clientIdentityKey interface{}) (unary grpc.U
 		if !ok {
 			panic("peer.FromContext expected to return a peer in grpc.UnaryServerInterceptor")
 		}
-		logger.WithField("peer_addr", fmt.Sprintf("%s", p.Addr)).Debug("peer addr")
+		logger.WithField("peer_addr", p.Addr.String()).Debug("peer addr")
 		a, ok := p.AuthInfo.(*authConnAuthType)
 		if !ok {
 			panic(fmt.Sprintf("NewInterceptors must be used in combination with grpc.NewTransportCredentials, but got auth type %T", p.AuthInfo))

rpc/grpcclientidentity/example/main.go

@@ -88,6 +88,9 @@ func server() {
 	log := logger.NewStderrDebugLogger()
 
 	srv, serve, err := grpchelper.NewServer(authListenerFactory, clientIdentityKey, log)
+	if err != nil {
+		onErr(err, "new server")
+	}
 	svc := &greeter{"hello "}
 	pdu.RegisterGreeterServer(srv, svc)
 

rpc/netadaptor/authlistener_netlistener_adaptor.go

@@ -25,8 +25,9 @@ package netadaptor
 import (
 	"context"
 	"fmt"
-	"github.com/zrepl/zrepl/logger"
 	"net"
+
+	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/transport"
 )
 

rpc/rpc_client.go

@@ -12,6 +12,7 @@ import (
 	"google.golang.org/grpc"
 
 	"github.com/google/uuid"
+
 	"github.com/zrepl/zrepl/replication/logic"
 	"github.com/zrepl/zrepl/replication/logic/pdu"
 	"github.com/zrepl/zrepl/rpc/dataconn"
@@ -85,7 +86,7 @@ func (c *Client) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, zfs.St
 	// TODO the returned sendStream may return a read error created by the remote side
 	res, streamCopier, err := c.dataClient.ReqSend(ctx, r)
 	if err != nil {
-		return nil, nil, nil
+		return nil, nil, err
 	}
 	if streamCopier == nil {
 		return res, nil, nil

rpc/rpc_debug.go

@@ -13,6 +13,7 @@ func init() {
 	}
 }
 
+//nolint[:deadcode,unused]
 func debug(format string, args ...interface{}) {
 	if debugEnabled {
 		fmt.Fprintf(os.Stderr, "rpc: %s\n", fmt.Sprintf(format, args...))

rpc/rpc_doc.go

@@ -115,4 +115,3 @@ package rpc
 // - remove the comments //
 // - vim: set virtualedit+=all
 // - vim: set ft=text
-

rpc/rpc_logging.go

@@ -12,9 +12,6 @@ type contextKey int
 
 const (
 	contextKeyLoggers contextKey = iota
-	contextKeyGeneralLogger
-	contextKeyControlLogger
-	contextKeyDataLogger
 )
 
 /// All fields must be non-nil

rpc/rpc_mux.go

@@ -4,8 +4,8 @@ import (
 	"context"
 	"time"
 
-	"github.com/zrepl/zrepl/transport"
 	"github.com/zrepl/zrepl/rpc/transportmux"
+	"github.com/zrepl/zrepl/transport"
 	"github.com/zrepl/zrepl/util/envconst"
 )
 

rpc/rpc_server.go

@@ -34,8 +34,6 @@ type Server struct {
 	dataServerServe    serveFunc
 }
 
-type serverContextKey int
-
 type HandlerContextInterceptor func(ctx context.Context) context.Context
 
 // config must be valid (use its Validate function).

rpc/transportmux/transportmux.go

@@ -7,6 +7,7 @@ package transportmux
 
 import (
 	"context"
+
 	"fmt"
 	"io"
 	"net"
@@ -49,7 +50,7 @@ func (l *demuxListener) Accept(ctx context.Context) (*transport.AuthConn, error)
 	return res.conn, res.err
 }
 
-type demuxAddr struct {}
+type demuxAddr struct{}
 
 func (demuxAddr) Network() string { return "demux" }
 func (demuxAddr) String() string  { return "demux" }
@@ -64,7 +65,7 @@ func (l *demuxListener) Close() error { return nil } // TODO
 // This is a protocol constant, changing it breaks the wire protocol.
 const LabelLen = 64
 
-func padLabel(out []byte, label string) (error) {
+func padLabel(out []byte, label string) error {
 	if len(label) > LabelLen {
 		return fmt.Errorf("label %q exceeds max length (is %d, max %d)", label, len(label), LabelLen)
 	}
@@ -142,7 +143,10 @@ func Demux(ctx context.Context, rawListener transport.AuthenticatedListener, lab
 				continue
 			}
 
-			rawConn.SetDeadline(time.Time{})
+			err = rawConn.SetDeadline(time.Time{})
+			if err != nil {
+				getLog(ctx).WithError(err).Error("cannot reset deadline")
+			}
 			// blocking is intentional
 			demuxListener.conns <- acceptRes{conn: rawConn, err: nil}
 		}
@@ -169,7 +173,12 @@ func (c labeledConnecter) Connect(ctx context.Context) (transport.Wire, error) {
 	}
 
 	if dl, ok := ctx.Deadline(); ok {
-		defer conn.SetDeadline(time.Time{})
+		defer func() {
+			err := conn.SetDeadline(time.Time{})
+			if err != nil {
+				getLog(ctx).WithError(err).Error("cannot reset deadline")
+			}
+		}()
 		if err := conn.SetDeadline(dl); err != nil {
 			closeConn(err)
 			return nil, err
@@ -202,4 +211,3 @@ func MuxConnecter(rawConnecter transport.Connecter, labels []string, timeout tim
 	}
 	return ret, nil
 }
-

rpc/versionhandshake/versionhandshake.go

@@ -39,7 +39,7 @@ func (e HandshakeError) Temporary() bool {
 	if e.isAcceptError {
 		return true
 	}
-	te, ok := e.IOError.(interface{ Temporary() bool });
+	te, ok := e.IOError.(interface{ Temporary() bool })
 	return ok && te.Temporary()
 }
 
@@ -52,11 +52,11 @@ func (e HandshakeError) Timeout() bool {
 	return false
 }
 
-func hsErr(format string, args... interface{}) *HandshakeError {
+func hsErr(format string, args ...interface{}) *HandshakeError {
 	return &HandshakeError{msg: fmt.Sprintf(format, args...)}
 }
 
-func hsIOErr(err error, format string, args... interface{}) *HandshakeError {
+func hsIOErr(err error, format string, args ...interface{}) *HandshakeError {
 	return &HandshakeError{IOError: err, msg: fmt.Sprintf(format, args...)}
 }
 
@@ -145,7 +145,7 @@ func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
 	if exts[len(exts)-1] != "" {
 		return hsErr("unexpected data trailing after last extension newline")
 	}
-	m.Extensions = exts[0:len(exts)-1]
+	m.Extensions = exts[0 : len(exts)-1]
 
 	return nil
 }
@@ -157,7 +157,7 @@ func DoHandshakeCurrentVersion(conn net.Conn, deadline time.Time) *HandshakeErro
 
 const HandshakeMessageMaxLen = 16 * 4096
 
-func DoHandshakeVersion(conn net.Conn, deadline time.Time, version int) *HandshakeError {
+func DoHandshakeVersion(conn net.Conn, deadline time.Time, version int) (rErr *HandshakeError) {
 	ours := HandshakeMessage{
 		ProtocolVersion: version,
 		Extensions:      nil,
@@ -167,8 +167,19 @@ func DoHandshakeVersion(conn net.Conn, deadline time.Time, version int) *Handsha
 		return hsErr("could not encode protocol banner: %s", err)
 	}
 
-	defer conn.SetDeadline(time.Time{})
+	err = conn.SetDeadline(deadline)
-	conn.SetDeadline(deadline)
+	if err != nil {
+		return hsErr("could not set deadline for protocol banner handshake: %s", err)
+	}
+	defer func() {
+		if rErr != nil {
+			return
+		}
+		err := conn.SetDeadline(time.Time{})
+		if err != nil {
+			rErr = hsErr("could not reset deadline after protocol banner handshake: %s", err)
+		}
+	}()
 	_, err = io.Copy(conn, bytes.NewBuffer(hsb))
 	if err != nil {
 		return hsErr("could not send protocol banner: %s", err)

rpc/versionhandshake/versionhandshake_test.go

@@ -3,13 +3,15 @@ package versionhandshake
 import (
 	"bytes"
 	"fmt"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/util/socketpair"
 	"io"
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/zrepl/zrepl/util/socketpair"
 )
 
 func TestHandshakeMessage_Encode(t *testing.T) {
@@ -23,8 +25,6 @@ func TestHandshakeMessage_Encode(t *testing.T) {
 	enc := string(encB)
 	t.Logf("enc: %s", enc)
 
-
-
 	assert.False(t, strings.ContainsAny(enc[0:10], " "))
 	assert.True(t, enc[10] == ' ')
 
@@ -68,7 +68,7 @@ func TestHandshakeMessage_DecodeReader(t *testing.T) {
 	require.NoError(t, err)
 
 	out := HandshakeMessage{}
-	err = out.DecodeReader(bytes.NewReader([]byte(enc)), 4 * 4096)
+	err = out.DecodeReader(bytes.NewReader([]byte(enc)), 4*4096)
 	assert.NoError(t, err)
 	assert.Equal(t, 2342, out.ProtocolVersion)
 	assert.Equal(t, 2, len(out.Extensions))

rpc/versionhandshake/versionhandshake_transport_wrappers.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"net"
 	"time"
+
 	"github.com/zrepl/zrepl/transport"
 )
 
@@ -41,7 +42,7 @@ type HandshakeListener struct {
 	timeout time.Duration
 }
 
-func (l HandshakeListener) Addr() (net.Addr) { return l.l.Addr() }
+func (l HandshakeListener) Addr() net.Addr { return l.l.Addr() }
 
 func (l HandshakeListener) Close() error { return l.l.Close() }
 

tlsconf/tlsconf.go

@@ -80,7 +80,9 @@ func (l *ClientAuthListener) Accept() (tcpConn *net.TCPConn, tlsConn *tls.Conn,
 	if err = tlsConn.Handshake(); err != nil {
 		goto CloseAndErr
 	}
-	tlsConn.SetDeadline(time.Time{})
+	if err = tlsConn.SetDeadline(time.Time{}); err != nil {
+		goto CloseAndErr
+	}
 
 	peerCerts = tlsConn.ConnectionState().PeerCertificates
 	if len(peerCerts) < 1 {