move implementation to internal/ directory (#828)

internal/replication/design.md

@@ -0,0 +1,367 @@
+The goal of this document is to describe what **logical steps** zrepl takes to replicate ZFS filesystems.
+Note that the actual code executing these steps is spread over the `endpoint.{Sender,Receiver}` RPC endpoint implementations. The code in `replication/{driver,logic}` (mostly `logic.Step.doReplication`) drives the replication and invokes the `endpoint` methods (locally or via RPC).
+Hence, when trying to map algorithm to implementation, use the code in package `replication` as the entrypoint and follow the RPC calls.
+
+
+* The [Single Replication Step](#zrepl-algo-single-step) algorithm is implemented as described
+  * step holds are implemented as described
+  * step bookmarks rely on bookmark cloning, which is WIP in OpenZFS (see respective TODO comments)
+  * the feature to hold receive-side `to` immediately after replication is used to move the `last-received-hold` forward
+    * this is a little more than what we allow in the algorithm (we only allow specifying a hold tag whereas moving the `last-received-hold` is a little more complex)
+  * the algorithm's sender-side callback is used move the `zrepl_CURSOR` bookmark to point to `to`
+* The `zrepl_CURSOR` bookmark and the `last-received-hold` ensure that future replication steps can always be done using incremental replication:
+  * Both reference / hold the last successfully received snapshot `to`.
+  * Thus, `to` or `#zrepl_CURSOR_G_${to.GUID}_J_${jobid}` can always be used for a future incremental replication step `to => future-to`:
+    * incremental send will be possible because `to` doesn't go away because we claim ownership of the `#zrepl_CURSOR` namespace
+    * incremental recv will be possible iff
+      * `to` will still be present on the recv-side because of `last-received-hold`
+      * the recv-side fs doesn't get newer snapshots than `to` in the meantime
+        * guaranteed because the zrepl model of the receiver assumes ownership of the filesystems it receives into
+          * if that assumption is broken, future replication attempts will fail with a conflict
+* The [Algorithm for Planning and Executing Replication of an Filesystems](#zrepl-algo-filesystem) is a design draft and not used.
+  However, there were some noteworthy lessons learned when implementing the algorithm for a single step:
+  * In order to avoid leaking `step-hold`s and `step-bookmarks`, if the replication planner is invoked a second time after a replication step (either initial or incremental) has been attempted but failed to completed, the replication planner must
+    * A) either guarantee that it will resume that replication step, and continue as if nothing happened or
+    * B) release the step holds and bookmarks and clear the partially received state on the sending side.
+  * Option A is what we want to do: we use the step algorithm to achieve resumability in the first place!
+  * Option B is not done by zrepl except if the sending side doesn't support resuming.
+    In that case however, we need not release any holds since the behavior is to re-start the send
+    from the beginning.
+  * However, there is one **edge-case to Option A for initial replication**:
+    * If initial replication "`full a`" fails without leaving resumable state, the step holds on the sending side are still present, which makes sense because
+      * a) we want resumability and
+      * b) **the sending side cannot immediately be informed post-failure whether the initial replication left any state that would mandate keeping the step hold**, because the network connection might have failed.
+    * Thus, the sending side must keep the step hold for "`full a`" until it knows more.
+    * In the current implementation, it knows more when the next replication attempt is made, the planner is invoked, the diffing algorithm run, and the `HintMostRecentCommonAncestor` RPC is sent by the active side, communicating the most recent common version shared betwen sender and receiver.
+    * At this point, the sender can safely throw away any step holds with CreateTXG's older than that version.
+  * **The `step-hold`, `step-bookmark`, `last-received-hold` and `replication-cursor` abstractions are currently local concepts of package `endpoint` and not part of the replication protocol**
+    * This is not necessarilty the best design decision and should be revisited some point:
+      * The (quite expensive) `HintMostRecentCommonAncestor` RPC impl on the sender would not be necessary if step holds were part of the replication protocol:
+        * We only need the `HintMostRecentCommonAncestor` info for the aforementioned edge-case during initial replication, where the receive is aborted without any partial received state being stored on the receiver (due to network failure, wrong zfs invocation, bad permissions, etc):
+        * **The replication planner does not know about the step holds, thus it cannot deterministically pick up where it left of (right at the start of the last failing initial replication).**
+        * Instead, it will seem like no prior invocation happened at all, and it will apply its policy for initial replication to pick a new `full b != full a`, **thereby leaking the step holds of `full a`**.
+        * In contrast, if the replication planner created the step holds and knew about them, it could use the step holds as an indicator where it left off and re-start from there (of course asserting that the thereby inferred step is compatible with the state of the receiving side).
+      * (What we do in zrepl right now is to hard-code the initial replication policy, and hard-code that assumption in `endpoint.ListStale` as well.)
+      * The cummulative cleanup done in `HintMostRecentCommonAncestor` provides a nice self-healing aspect, though.
+
+* We also have [Notes on Planning and Executing Replication of Multiple Filesystems](#zrepl-algo-multiple-filesystems-notes)
+
+---
+
+<a id="zrepl-algo-single-step"></a>
+## Algorithm for a Single Replication Step
+
+The algorithm described below describes how a single replication step must be implemented.
+A *replication step* is a full send of a snapshot `to` for initial replication, or an incremental send (`from => to`) for incremental replication.
+
+The algorithm **ensures resumability** of the replication step in presence of
+
+* uncoordinated or unsynchronized destruction of the filesystem, snapshots, or bookmarks involved in the replication step
+* network failures at any time
+* other instances of this algorithm executing the same step in parallel (e.g. concurrent replication to different destinations)
+
+To accomplish this goal, the algorithm **assumes ownership of parts of the ZFS hold tag namespace and the bookmark namespace**:
+* holds with prefix `zrepl_STEP` on any snapshot are reserved for zrepl
+* bookmarks with prefix `zrepl_STEP` are reserved for zrepl
+
+Resumability of a step cannot be guaranteed if these sub-namespaces are manipulated through software other than zrepl.
+
+Note that the algorithm **does not ensure** that a replication *plan*, which describes a *set* of replication steps, can be carried out successfully.
+If that is desirable, additional measures outside of this algorithm must be taken.
+
+---
+
+### Definitions:
+
+#### Step Completion & Invariants
+
+The replication step (full `to` send or `from => to` send) is *complete* iff the algorithm ran to completion without errors or a permanent non-network error is reported by sender or receiver.
+
+Specifically, the algorithm may be invoked with the same `from` and `to` arguments, and potentially a `resume_token`, after a temporary (like network-related) failure:
+**Unless permanent errors occur, repeated invocations of the algorithm with updated resume token will converge monotonically (but not strictly monotonically) toward completion.**
+
+Note that the mere existence of `to` on the receiving side does not constitute completion, since there may still be post-recv actions to be performed on sender and receiver.
+
+#### Job and Job ID
+This algorithm supports that *multiple* instance of it run in parallel on the *same* step (full `to` / `from => to` pair).
+An exemplary use case for this feature are concurrent replication jobs that replicate the same dataset to different receivers.
+
+**We require that all parallel invocations of this algorithm provide different and unique `jobid`s.**
+Violation of `jobid` uniqueness across parallel jobs may result in interference between instances of this algorithm, resulting in potential compromise of resumability.
+
+After a step is *completed*, `jobid` is guaranteed to not be encoded in on-disk state.
+Before a step is completed, there is no such guarantee.
+Changing the `jobid` before step completion may compromise resumability and may leak the underlying ZFS holds or step bookmarks (i.e. zrepl won't clean them up)
+Note the definition of *complete* above.
+
+#### Step Bookmark
+
+A step bookmark is our equivalent of ZFS holds, but for bookmarks.<br/>
+A step bookmark is a ZFS bookmark whose name matches the following regex:
+
+```
+STEP_BOOKMARK = #zrepl_STEP_bm_G_([0-9a-f]+)_J_(.+)
+```
+
+- capture group `1` must be the guid of `zfs get guid ${STEP_BOOKMARK}`, encoded hexadecimal (without leading `0x`) and fixed-length (i.e. padded with leading zeroes)
+- capture group `2` must be equal to the `jobid`
+
+### Algorithm
+
+INPUT:
+
+* `jobid`
+* `from`: snapshot or bookmark: may be nil for full send
+* `to`: snapshot, must never be nil
+* `resume_token` (may be nil)
+* (`from` and `to` must be on the same filesystem)
+
+#### Prepare-Phase
+
+Send-side: make sure `to` and `from` don't go away
+
+- hold `to` using `idempotent_hold(to, zrepl_STEP_J_${jobid})`
+- make sure `from` doesn't go away:
+  - if `from` is a snapshot: hold `from` using `idempotent_hold(from, zrepl_STEP_J_${jobid})`
+  - else `idempotent_step_bookmark(from)` (`from` is a bookmark)
+    - PERMANENT ERROR if this fails (e.g. because `from` is a bookmark whose snapshot no longer exists and we don't have bookmark copying yet (ZFS feature is in development) )
+      - Why? we must assume the given bookmark is externally managed (i.e. not in the )
+      - this means the bookmark is externally created bookmark and cannot be trusted to persist until the replication step succeeds
+      - Maybe provide an override-knob for this behavior
+
+Recv-side: no-op
+
+- `from` cannot go away once we received enough data for the step to be resumable:
+    ```text
+    # do a partial incremental recv @a=>@b (i.e. recv with -s, and abort the incremental recv in the middle)
+    # try destroying the incremental source on the receiving side
+    zfs destroy p1/recv@a
+    cannot destroy 'p1/recv@a': snapshot has dependent clones
+    use '-R' to destroy the following datasets:
+    p1/recv/%recv
+    # => doesn't work, because zfs recv is implemented as a `clone` internally, that's exactly what we want
+    ```
+
+- if recv-side `to` exists, goto cleanup-phase (no replication to do)
+  
+- `to` cannot be destroyed while being received, because it isn't visible as a snapshot yet (it isn't yet one after all)
+
+#### Replication Phase
+
+Attempt the replication step:
+start `zfs send` and pipe its output into `zfs recv` until an error occurs or `recv` returns without error.
+
+Let us now think about interferences that could occur during this phase, and ensure that none of them compromise the goals of this algorithm, i.e., monotonic convergence toward step completion using resumable send & recv.
+
+**Safety from External Pruning**
+We are safe from pruning during the replication step because we have guarantees that no external action will destroy send-side `from` and `to`, and recv-side `to` (for both snapshot and bookmark `from`s)<br/>
+
+**Safety In Presence of Network Failures During Replication**
+Network failures during replication can be recovered from using resumable send & recv:
+
+- Network failure before the receive-side could stored data:
+  - Send-side `from` and `to` are guaranteed to be still present due to zfs holds
+  - recv-side `from` may no longer exist because we don't `hold` it explicitly
+    - if that is the case, ERROR OUT, step can never complete
+  - If the step planning algorithm does not include the step, for example because a snapshot filter configuration was changed by the user inbetween which hides `from` or `to` from the second planning attempt: **tough luck, we're leaking all holds**
+- Network failure during the replication
+  - send-side `from` and `to` are still present due to zfs holds
+  - recv-side `from` is still present because the partial receive state prevents its destruction (see prepare-phase)
+  - if recv-side has a resume token, the resume token will continue to work on the sender because `from`s and `to` are still present
+- Network failure at the end of the replication step stream transmission
+  - Variant A: failure from the sender's perspective, success from the receiver's perspective
+    - receive-side `to` doesn't have a hold and could be destroyed anytime
+    - receive-side `from` doesn't have a hold and could be destroyed anytime
+    - thus, when the step is invoked again, pattern match `(from_exists, to_exists)`
+      - `(true, true)`: prepare-phase will early-exit
+      - `(false,true)`: prepare-phase will error out bc. `from` does not exist
+      - `(true, false)`: entire step will be re-done
+        - FIXME monotonicity requirement does not hold
+      - `(false, false)`: prepare-phase will error out bc. `from` does not exist
+  - Variant B: success from the sender's perspective, failure from the receiver's perspective
+    - No idea how this would happen except for bugs in error reporting in the replication protocol
+    - Misclassification by the sender, most likely broken error handling in the sender or replication protocol
+    - => the sender will release holds and move the replication cursor while the receiver won't => tough luck
+
+If the RPC used for `zfs recv` returned without error, this phase is done.
+(Although the *replication step* (this algorithm) is not yet *complete*, see definition of complete).
+
+
+#### Cleanup-Phase
+
+At the end of this phase, all intermediate state we built up to support the resumable replication of the step is destroyed.
+However, consumers of this algorithm might want to take advantage of the fact that we currently still have holds / step bookmarks.
+
+##### Recv-side: Optionally hold `to` with a caller-defined tag
+
+Users of the algorithm might want to depend on `to` being available on the receiving side for a longer duration than the lifetime of the current step's algorithm invocation.
+For reasons explained in the next paragraph, we cannot guarantee that we have a `hold` when `recv` exists. If that were the case, we could take our time to provide a generalized callback to the user of the algorithm, and have them do whatever they want with `to` while we guarantee that `to` doesn't go away through the hold. But that functionality isn't available, so we only provide a fixed operation right after receive: **take a `hold` with a tag of the algorithm user's choice**. That's what's needed to guarantee that a replication plan, consisting of multiple consecutive steps, can be carried out without a receive-side prune job interfering by destroying `to`. Technically, this step is racy, i.e., it could happen that `to` is destroyed between `recv` and `hold`. But this is a) unlikely and b) not fatal because we can detect that hold fails because the 'dataset does not exist` and re-do the entire transmission since we still hold send-side `from` and `to`, i.e. we just lose a lot of work in rare cases.
+
+So why can't we have a `hold` at the time `recv` exits?
+It seems like [`zfs send --holds`](https://github.com/zfsonlinux/zfs/commit/9c5e88b1ded19cb4b19b9d767d5c71b34c189540) could be used to send the send-side's holds to the receive-side. But that flag always sends all holds, and `--holds` is implemented in user-space libzfs, i.e., doesn't happen in the same txg as the recv completion. Thus, the race window is in fact only smaller (unless we oversaw some synchronization in userland zfs).
+But if we're willing to entertain the idea a little further, we still hit the problem that `--holds` sends _all_ holds, whereas our goal is to _temporarily_ have >= 1 hold that we own until the callback is done, and then release all of the received holds so that no holds created by us exist after this algorithm completes.
+Since there could be concurrent `zfs hold` invocations on the sender and receiver while this algorithm runs, and because `--holds` doesn't provide info about which holds were sent, we cannot correctly destroy _exactly_ those holds that we received.
+
+##### Send-side: callback to consumer, then cleanup
+We can provide the algorithm user with a generic callback because we have holds / step bookmarks for `to` and `from`, respectively.
+
+Example use case for the sender-side callback is the replication cursor, see algorithm for filesystem replication below.
+
+After the callback is done: cleanup holds & step bookmark:
+
+- `idempotent_release(to, zrepl_STEP_J_${jobid})`
+- make sure `from` can now go away:
+  - if `from` is a snapshot: `idempotent_release(from, zrepl_STEP_J_${jobid})`
+  - else `idempotent_step_bookmark_destroy(from)`
+    - if `from` fulfills the properties of a step bookmark: destroy it
+    - otherwise: it's a bookmark not created by this algorithm that happened to be used for replication: leave it alone
+      - that can only happen if user used the override knob
+
+Note that "make sure `from` can now go away" is the inverse of "Send-side: make sure `to` and `from` don't go away". Those are relatively complex operations and should be implemented in the same file next to each other to ease maintainability.
+
+---
+
+### Notes & Pseudo-APIs used in the algorithm
+
+- `idempotent_hold(snapshot s, string tag)` like zfs hold, but doesn't error if hold already exists
+- `idempotent_release(snapshot s, string tag)` like zfs hold, but doesn't error if hold already exists
+- `idempotent_step_bookmark(snapshot_or_bookmark b)` creates a *step bookmark* of b
+  - determine step bookmark name N
+  - if `N` already exists, verify it's a correct step bookmark => DONE or ERROR
+  - if `b` is a snapshot, issue `zfs bookmark`
+  - if `b` is a bookmark:
+    - if bookmark cloning supported, use it to duplicate the bookmark
+    - else ERROR OUT, with an error that upper layers can identify as such, so that they are able to ignore the fact that we couldn't create a step bookmark 
+- `idempotent_destroy(bookmark #b_$GUID, of a snapshot s)` must atomically check that `$GUID == s.guid` before destroying s
+- `idempotent_bookmark(snapshot s, $GUID, name #b_$GUID)` must atomically check that `$GUID == s.guid` at the time of bookmark creation
+- `idempotent_destroy(snapshot s)` must atomically check that zrepl's `s.guid` matches the current `guid` of the snapshot (i.e. destroy by guid)
+
+
+<a id="zrepl-algo-filesystem"></a>
+## Algorithm for Planning and Executing Replication of a Filesystems (planned, not implemented yet)
+
+This algorithm describes how a filesystem or zvol is replicated in zrepl.
+The algorithm is invoked with a `jobid`, `sender`, `receiver`, a sender-side `filesystem path`.
+It builds a diff between the sender and receiver filesystem bookmarks+snapshots and determines whether a replication conflict exists or whether fast-forward replication using full or incremental sends is possible.
+In case of conflict, the algorithm errors out with a conflict description that can be used to manually or automatically resolve the conflict.
+Otherwise, the algorithm builds a list of replication steps that are then worked on sequentially by the "Algorithm for a Single Replication Step".
+
+The algorithm ensures that a plan can be executed exactly as planned by acquiring appropriate zfs holds.
+The algorithm can be configured to retry a plan when encountering non-permanent errors (e.g. network errors).
+However, permanent errors result in the plan being cancelled.
+
+Regardless of whether a plan can be executed to completion or is cancelled, the algorithm guarantees that leftover artifacts (i.e. holds) of its invocation are cleaned up.
+However, since network failure can occur at any point, there might be stale holds on sending or receiving side after a crash or other error.
+These will be cleaned up on a subsequent invocation of this algorithm with the same `jobid`.
+
+The algorithm is fit to be executed in parallel from the same sender to different receivers.
+To be clear: replicating in parallel from the same sender to different receivers is supported.
+But one instance of the algorithm assumes ownership of the `filesystem path` on the receiver.
+
+The algorithm reserves the following sub-namespaces:
+* zfs hold: `zrepl_FS`
+* bookmark names: `zrepl_FS`
+
+### Definitions
+
+#### ZFS Filesystem Path
+We refer to the name of a ZFS filesystem or volume as a *filesystem path*, sometimes just *path*.
+The name includes the pool name.
+
+#### Sender and Receiver
+Sender and Receiver are passed as RPC stubs to the algorithm.
+One of those RPC stubs will typically be local, i.e., call methods on a struct in the same address space.
+The other RPC stub may invoke actual calls over the network (unless local replication is used).
+The algorithm does not make any assumption about which object is local or an actual stub.
+This design decouples the replication logic (described in this algorithm) from the question which side (sender or receiver) initiates the replication.
+
+### Algorithm
+
+**Cleanup Previous Invocations With Same `jobid`** by scanning the filesystem's list of snapshots and step bookmarks, and destroying any which was created by this algorithm when it was invoked with `jobid`.<br/>
+TODO HOW
+
+**Build a fast-forward list of replication steps** `STEPS`.
+`STEPS` may contain an optional initial full send, and subsequent incremental send steps.
+`STEPS[0]` may also have a resume token.
+If fast-forward is not possible, produce a conflict description and ERROR OUT.<br/>
+TODOs:
+- make it configurable what snapshots are included in the list (i.e. every one we see, only most recent, at least one every X hours, ...)
+
+**Ensure that we will be able to carry out all steps** by acquiring holds or fsstep bookmarks on the sending side
+- `idempotent_hold([s.to for s in STEPS], zrepl_FS_J_${jobid})`
+- `if STEPS[0].from != nil: idempotent_FSSTEP_bookmark(STEPS[0].from, zrepl_FSSTEP_bm_G_${STEPS[0].from.guid}_J_${jobid})` 
+  
+**Determine which steps have not been completed (`uncompleted_steps`)** (we might be in an second invocation of this algorithm after a network failure and some steps might already be done):
+- `res_tok := receiver.ResumeToken(fs)`
+- `rmrfsv := receiver.MostRecentFilesystemVersion(fs)`
+- if `res_tok !=  nil`: ensure that `res_tok` has a corresponding step in `STEPS`, otherwise ERROR OUT
+- if `rmrfsv != nil`: ensure that `res_tok` has a corresponding step in `STEPS`, otherwise ERROR OUT
+- if `(res_token != nil && rmrfsv != nil)`: ensure that `res_tok` is the subsequent step to the one we found for `rmrfsv`
+- if both are nil, we are at the beginning, `uncompleted_steps = STEPS` and goto next block
+- `rstep := if res_tok != nil { res_tok } else { rmrfsv }`
+- `uncompleted_steps := STEPS[find_step_idx(STEPS, rstep).expect("must exist, checked above"):]`
+  - Note that we do not explicitly check for the completion of prior replication steps.
+    All we care about is what needs to be done from `rstep`.
+  - This is intentional and necessary because we cumulatively release all holds and step bookmarks made for steps that precede a just-completed step (see next paragraph)
+
+**Execute uncompleted steps**<br/>
+Invoke the "Algorithm for a Single Replication Step" for each step in `uncompleted_steps`.
+Remember to pass the resume token if one exists.
+
+In the wind-down phase of each replication step `from => to`, while the per-step algorithm still holds send-side `from` and `to`, as well as recv-side `to`:
+
+- Sending side: **Idempotently move the replication cursor to `to`.**
+  - Why: replication cursor tracks the sender's last known replication state, outlives the current plan, but is required to guarantee that future invocations of this algorithm find an incremental path.
+  - Impl for 'atomic' move: have two cursors (fixes [#177](https://github.com/zrepl/zrepl/issues/177))
+      - Idempotently cursor-bookmark `to` using `idempotent_bookmark(to, to.guid, #zrepl_replication_cursor_G_${to.guid}_J_${jobid})`
+      - Idempotently destroy old cursor-bookmark of `from` `idempotent_destroy(#zrepl_replication_cursor_G_${from.guid}_J_${jobid}, from)`
+    - If `from` is a snapshot, release the hold on it using `idempotent_release(from,  zrepl_J_${jobid})`
+    - FIXME: resumability if we crash inbetween those operations (or scrap it and wait for channel programs to bring atomicity)
+
+- Receiving side: **`idempotent_hold(to, zrepl_FS_J_${jobid})` because its going to be the next step's `from`**
+  - As discussed in the section on the per-step algorithm, this is a feature provided to us by the per-step algorithm.
+
+- Receiving side + Sending side (order doesn't matter): Make sure all holds & step bookmarks made by this plan on already replicated snapshots are released / destroyed:
+  - `idempotent_release_prior_and_including(from, zrepl_FS_TODO)`
+  - `idempotent_step_bookmark_destroy_prior_and_including(from, zrepl_FS_TODO)`
+
+**Cleanup** receiving and sending side (order doesn't matter)
+
+  - `idempotent_release_prior_and_including(STEPS[-1].to, zrepl_FS_TODO)`
+  - `idempotent_step_bookmark_destroy_prior_and_including(STEPS[-1].from, zrepl_FS_TODO)`
+
+### Notes
+
+- `idempotent_FSSTEP_bookmark` is like `idempotent_STEP_bookmark`, but with prefix `zrepl_FSSTEP` instead of `zrepl_STEP`
+
+
+<a id="zrepl-algo-multiple-filesystems-notes"></a>
+
+## Notes on Planning and Executing Replication of Multiple Filesystems
+
+### The RPC Interfaces Uses Send-side Filesystem Names to Identify a Filesystem
+
+The model of the receiver includes the assumption that it will `Receive` all snapshot streams sent to it into filesystems with paths prefixed with `root_fs`.
+This is useful for separating filesystem path namespaces for multiple clients.
+The receiver hides this prefixing in its RPC interface, i.e., when responding to `ListFilesystems` rpcs, the prefix is removed before sending the response.
+This behavior is useful to achieve symmetry in this algorithm: we do not need to take the prefixing into consideration when computing diffs.
+For the receiver, it has the advantage that `Receive` rpc can enforce the namespacing and need not trust this algorithm to apply it correctly.
+
+The receiver is also allowed (and may need to) do implement other filtering / namespace transformations.
+For example, when only `foo/a/b` has been received, but not `foo` nor `foo/a`, the receiver must not include the latter two in its `ListFilesystems` response.
+The current zrepl receiver endpoint implementation uses the `zrepl:placeholder` property to mark filesystems `foo` and `foo/a` as placeholders, and filters out such filesystems in the `ListFilesystems` response.
+Another approach would be to have a flat list of received filesystems per sender, and have a separate table that associates send-side names and receive-side filesystem paths.
+
+Similarly, the sender is allowed to filter the output of its RPC interface to hide certain filesystems or snapshots.
+
+#### Consequences of the Above Design
+
+Namespace filtering and transformations of filesystem paths on sender and receiver are user-configurable.
+Both ends of the replication setup have their own config, and do not coordinate config changes.
+This results in a number of challenges:
+
+- A sender might change its filter to allow additional filesystems to be replicated.
+  For example, where `foo/a/b` was initially allowed, the new filter might allow `foo` and `foo/a` as well.
+  If the receiver uses the `zrepl:placeholder` approach as sketched out above, this means that the receiver will need to replace the placeholder filesystems `$root_fs/foo` and `$root_fs/foo/a` with the incoming full sends of `foo` and `foo/a`.
+
+- Send-side renames cannot be handled efficiently because send-side rename effectively changes the filesystem identity because we use its name.

internal/replication/driver/errorclass_enumer.go

@@ -0,0 +1,49 @@
+// Code generated by "enumer -type=errorClass"; DO NOT EDIT.
+
+package driver
+
+import (
+	"fmt"
+)
+
+const _errorClassName = "errorClassPermanenterrorClassTemporaryConnectivityRelated"
+
+var _errorClassIndex = [...]uint8{0, 19, 57}
+
+func (i errorClass) String() string {
+	if i < 0 || i >= errorClass(len(_errorClassIndex)-1) {
+		return fmt.Sprintf("errorClass(%d)", i)
+	}
+	return _errorClassName[_errorClassIndex[i]:_errorClassIndex[i+1]]
+}
+
+var _errorClassValues = []errorClass{0, 1}
+
+var _errorClassNameToValueMap = map[string]errorClass{
+	_errorClassName[0:19]:  0,
+	_errorClassName[19:57]: 1,
+}
+
+// errorClassString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func errorClassString(s string) (errorClass, error) {
+	if val, ok := _errorClassNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to errorClass values", s)
+}
+
+// errorClassValues returns all values of the enum
+func errorClassValues() []errorClass {
+	return _errorClassValues
+}
+
+// IsAerrorClass returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i errorClass) IsAerrorClass() bool {
+	for _, v := range _errorClassValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}

internal/replication/driver/replication_driver.go

@@ -0,0 +1,847 @@
+package driver
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/go-playground/validator"
+	"github.com/kr/pretty"
+	"github.com/pkg/errors"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/zrepl/zrepl/internal/daemon/logging/trace"
+	"github.com/zrepl/zrepl/internal/zfs"
+
+	"github.com/zrepl/zrepl/internal/replication/report"
+	"github.com/zrepl/zrepl/internal/util/chainlock"
+)
+
+type interval struct {
+	begin time.Time
+	end   time.Time
+}
+
+func (w *interval) SetZero() {
+	w.begin = time.Time{}
+	w.end = time.Time{}
+}
+
+// Duration of 0 means indefinite length
+func (w *interval) Set(begin time.Time, duration time.Duration) {
+	if begin.IsZero() {
+		panic("zero begin time now allowed")
+	}
+	w.begin = begin
+	w.end = begin.Add(duration)
+}
+
+// Returns the End of the interval if it has a defined length.
+// For indefinite lengths, returns the zero value.
+func (w *interval) End() time.Time {
+	return w.end
+}
+
+// Return a context with a deadline at the interval's end.
+// If the interval has indefinite length (duration 0 on Set), return ctx as is.
+// The returned context.CancelFunc can be called either way.
+func (w *interval) ContextWithDeadlineAtEnd(ctx context.Context) (context.Context, context.CancelFunc) {
+	if w.begin.IsZero() {
+		panic("must call Set before ContextWIthDeadlineAtEnd")
+	}
+	if w.end.IsZero() {
+		// indefinite length, just return context as is
+		return ctx, func() {}
+	} else {
+		return context.WithDeadline(ctx, w.end)
+	}
+}
+
+type run struct {
+	l *chainlock.L
+
+	startedAt, finishedAt time.Time
+
+	waitReconnect      interval
+	waitReconnectError *timedError
+
+	// the attempts attempted so far:
+	// All but the last in this slice must have finished with some errors.
+	// The last attempt may not be finished and may not have errors.
+	attempts []*attempt
+}
+
+type Planner interface {
+	Plan(context.Context) ([]FS, error)
+	WaitForConnectivity(context.Context) error
+}
+
+// an attempt represents a single planning & execution of fs replications
+type attempt struct {
+	planner Planner
+	config  Config
+
+	l *chainlock.L
+
+	startedAt, finishedAt time.Time
+
+	// after Planner.Plan was called, planErr and fss are mutually exclusive with regards to nil-ness
+	// if both are nil, it must be assumed that Planner.Plan is active
+	planErr *timedError
+	fss     []*fs
+}
+
+type timedError struct {
+	Err  error
+	Time time.Time
+}
+
+func newTimedError(err error, t time.Time) *timedError {
+	if err == nil {
+		panic("error must be non-nil")
+	}
+	if t.IsZero() {
+		panic("t must be non-zero")
+	}
+	return &timedError{err, t}
+}
+
+func (e *timedError) IntoReportError() *report.TimedError {
+	if e == nil {
+		return nil
+	}
+	return report.NewTimedError(e.Err.Error(), e.Time)
+}
+
+type FS interface {
+	// Returns true if this FS and fs refer to the same filesystem returned
+	// by Planner.Plan in a previous attempt.
+	EqualToPreviousAttempt(fs FS) bool
+	// The returned steps are assumed to be dependent on exactly
+	// their direct predecessors in the returned list.
+	PlanFS(context.Context) ([]Step, error)
+	ReportInfo() *report.FilesystemInfo
+}
+
+type Step interface {
+	// Returns true iff the target snapshot is the same for this Step and other.
+	// We do not use TargetDate to avoid problems with wrong system time on
+	// snapshot creation.
+	//
+	// Implementations can assume that `other` is a step of the same filesystem,
+	// although maybe from a previous attempt.
+	// (`same` as defined by FS.EqualToPreviousAttempt)
+	//
+	// Note that TargetEquals should return true in a situation with one
+	// originally sent snapshot and a subsequent attempt's step that uses
+	// resumable send & recv.
+	TargetEquals(other Step) bool
+	TargetDate() time.Time
+	Step(context.Context) error
+	ReportInfo() *report.StepInfo
+}
+
+type fs struct {
+	fs FS
+
+	l *chainlock.L
+
+	blockedOn report.FsBlockedOn
+
+	// ordering relationship that must be maintained for initial replication
+	initialRepOrd struct {
+		parents, children []*fs
+		parentDidUpdate   chan struct{}
+	}
+
+	planning struct {
+		waitingForStepQueue bool
+		done                bool
+		err                 *timedError
+	}
+
+	// valid iff planning.done && planning.err == nil
+	planned struct {
+		// valid iff planning.done && planning.err == nil
+		stepErr *timedError
+		// all steps, in the order in which they must be completed
+		steps []*step
+		// index into steps, pointing at the step that is currently executing
+		// if step >= len(steps), no more work needs to be done
+		step int
+	}
+}
+
+type step struct {
+	l    *chainlock.L
+	step Step
+}
+
+type ReportFunc func() *report.Report
+type WaitFunc func(block bool) (done bool)
+
+type Config struct {
+	StepQueueConcurrency     int           `validate:"gte=1"`
+	MaxAttempts              int           `validate:"eq=-1|gt=0"`
+	ReconnectHardFailTimeout time.Duration `validate:"gt=0"`
+}
+
+var validate = validator.New()
+
+func (c Config) Validate() error {
+	return validate.Struct(c)
+}
+
+// caller must ensure config.Validate() == nil
+func Do(ctx context.Context, config Config, planner Planner) (ReportFunc, WaitFunc) {
+
+	if err := config.Validate(); err != nil {
+		panic(err)
+	}
+
+	log := getLog(ctx)
+	l := chainlock.New()
+	run := &run{
+		l:         l,
+		startedAt: time.Now(),
+	}
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+
+		defer run.l.Lock().Unlock()
+		log.Debug("begin run")
+		defer log.Debug("run ended")
+		var prev *attempt
+		mainLog := log
+		for ano := 0; ano < int(config.MaxAttempts); ano++ {
+			log := mainLog.WithField("attempt_number", ano)
+			log.Debug("start attempt")
+
+			run.waitReconnect.SetZero()
+			run.waitReconnectError = nil
+
+			// do current attempt
+			cur := &attempt{
+				l:         l,
+				startedAt: time.Now(),
+				planner:   planner,
+				config:    config,
+			}
+			run.attempts = append(run.attempts, cur)
+			run.l.DropWhile(func() {
+				cur.do(ctx, prev)
+			})
+			prev = cur
+			if ctx.Err() != nil {
+				log.WithError(ctx.Err()).Info("context error")
+				return
+			}
+
+			// error classification, bail out if done / permanent error
+			rep := cur.report()
+			log.WithField("attempt_state", rep.State).Debug("attempt state")
+			errRep := cur.errorReport()
+
+			if rep.State == report.AttemptDone {
+				if len(rep.Filesystems) == 0 {
+					log.Warn("no filesystems were considered for replication")
+				}
+				log.Debug("attempt completed")
+				break
+			}
+
+			mostRecentErr, mostRecentErrClass := errRep.MostRecent()
+			log.WithField("most_recent_err", mostRecentErr).WithField("most_recent_err_class", mostRecentErrClass).Debug("most recent error used for re-connect decision")
+			if mostRecentErr == nil {
+				// inconsistent reporting, let's bail out
+				log.WithField("attempt_state", rep.State).Warn("attempt does not report done but error report does not report errors, aborting run")
+				break
+			}
+			log.WithError(mostRecentErr.Err).Error("most recent error in this attempt")
+			shouldReconnect := mostRecentErrClass == errorClassTemporaryConnectivityRelated
+			log.WithField("reconnect_decision", shouldReconnect).Debug("reconnect decision made")
+			if shouldReconnect {
+				run.waitReconnect.Set(time.Now(), config.ReconnectHardFailTimeout)
+				log.WithField("deadline", run.waitReconnect.End()).Error("temporary connectivity-related error identified, start waiting for reconnect")
+				var connectErr error
+				var connectErrTime time.Time
+				run.l.DropWhile(func() {
+					ctx, cancel := run.waitReconnect.ContextWithDeadlineAtEnd(ctx)
+					defer cancel()
+					connectErr = planner.WaitForConnectivity(ctx)
+					connectErrTime = time.Now()
+				})
+				if connectErr == nil {
+					log.Error("reconnect successful") // same level as 'begin with reconnect' message above
+					continue
+				} else {
+					run.waitReconnectError = newTimedError(connectErr, connectErrTime)
+					log.WithError(connectErr).Error("reconnecting failed, aborting run")
+					break
+				}
+			} else {
+				log.Error("most recent error cannot be solved by reconnecting, aborting run")
+				return
+			}
+
+		}
+
+	}()
+
+	wait := func(block bool) bool {
+		if block {
+			<-done
+		}
+		select {
+		case <-done:
+			return true
+		default:
+			return false
+		}
+	}
+	report := func() *report.Report {
+		defer run.l.Lock().Unlock()
+		return run.report()
+	}
+	return report, wait
+}
+
+func (a *attempt) do(ctx context.Context, prev *attempt) {
+	prevs := a.doGlobalPlanning(ctx, prev)
+	if prevs == nil {
+		return
+	}
+	a.doFilesystems(ctx, prevs)
+}
+
+// if no error occurs, returns a map that maps this attempt's a.fss to `prev`'s a.fss
+func (a *attempt) doGlobalPlanning(ctx context.Context, prev *attempt) map[*fs]*fs {
+	ctx, endSpan := trace.WithSpan(ctx, "plan")
+	defer endSpan()
+	pfss, err := a.planner.Plan(ctx)
+	errTime := time.Now()
+	defer a.l.Lock().Unlock()
+	if err != nil {
+		a.planErr = newTimedError(err, errTime)
+		a.fss = nil
+		a.finishedAt = time.Now()
+		return nil
+	}
+
+	// a.fss != nil indicates that there was no planning error (see doc comment)
+	a.fss = make([]*fs, 0)
+
+	for _, pfs := range pfss {
+		fs := &fs{
+			fs:        pfs,
+			l:         a.l,
+			blockedOn: report.FsBlockedOnNothing,
+		}
+		fs.initialRepOrd.parentDidUpdate = make(chan struct{}, 1)
+		a.fss = append(a.fss, fs)
+	}
+
+	prevs := make(map[*fs]*fs)
+	{
+		prevFSs := make(map[*fs][]*fs, len(pfss))
+		if prev != nil {
+			debug("previous attempt has %d fss", len(a.fss))
+			for _, fs := range a.fss {
+				for _, prevFS := range prev.fss {
+					if fs.fs.EqualToPreviousAttempt(prevFS.fs) {
+						l := prevFSs[fs]
+						l = append(l, prevFS)
+						prevFSs[fs] = l
+					}
+				}
+			}
+		}
+		type inconsistency struct {
+			cur   *fs
+			prevs []*fs
+		}
+		var inconsistencies []inconsistency
+		for cur, fss := range prevFSs {
+			if len(fss) > 1 {
+				inconsistencies = append(inconsistencies, inconsistency{cur, fss})
+			}
+		}
+		sort.SliceStable(inconsistencies, func(i, j int) bool {
+			return inconsistencies[i].cur.fs.ReportInfo().Name < inconsistencies[j].cur.fs.ReportInfo().Name
+		})
+		if len(inconsistencies) > 0 {
+			var msg strings.Builder
+			msg.WriteString("cannot determine filesystem correspondences between different attempts:\n")
+			var inconsistencyLines []string
+			for _, i := range inconsistencies {
+				var prevNames []string
+				for _, prev := range i.prevs {
+					prevNames = append(prevNames, prev.fs.ReportInfo().Name)
+				}
+				l := fmt.Sprintf("  %s => %v", i.cur.fs.ReportInfo().Name, prevNames)
+				inconsistencyLines = append(inconsistencyLines, l)
+			}
+			fmt.Fprint(&msg, strings.Join(inconsistencyLines, "\n"))
+			now := time.Now()
+			a.planErr = newTimedError(errors.New(msg.String()), now)
+			a.fss = nil
+			a.finishedAt = now
+			return nil
+		}
+		for cur, fss := range prevFSs {
+			if len(fss) > 0 {
+				prevs[cur] = fss[0]
+			}
+		}
+	}
+	// invariant: prevs contains an entry for each unambiguous correspondence
+
+	// build up parent-child relationship (FIXME (O(n^2), but who's going to have that many filesystems...))
+	mustDatasetPathOrPlanFail := func(fs string) *zfs.DatasetPath {
+		dp, err := zfs.NewDatasetPath(fs)
+		if err != nil {
+			now := time.Now()
+			a.planErr = newTimedError(errors.Wrapf(err, "%q", fs), now)
+			a.fss = nil
+			a.finishedAt = now
+			return nil
+		}
+		return dp
+	}
+	for _, f1 := range a.fss {
+		fs1 := mustDatasetPathOrPlanFail(f1.fs.ReportInfo().Name)
+		if fs1 == nil {
+			return nil
+		}
+		for _, f2 := range a.fss {
+			fs2 := mustDatasetPathOrPlanFail(f2.fs.ReportInfo().Name)
+			if fs2 == nil {
+				return nil
+			}
+			if fs1.HasPrefix(fs2) && !fs1.Equal(fs2) {
+				f1.initialRepOrd.parents = append(f1.initialRepOrd.parents, f2)
+				f2.initialRepOrd.children = append(f2.initialRepOrd.children, f1)
+			}
+		}
+	}
+
+	return prevs
+}
+
+func (a *attempt) doFilesystems(ctx context.Context, prevs map[*fs]*fs) {
+	ctx, endSpan := trace.WithSpan(ctx, "do-repl")
+	defer endSpan()
+
+	defer a.l.Lock().Unlock()
+
+	stepQueue := newStepQueue()
+	defer stepQueue.Start(a.config.StepQueueConcurrency)()
+	var fssesDone sync.WaitGroup
+	for _, f := range a.fss {
+		fssesDone.Add(1)
+		go func(f *fs) {
+			defer fssesDone.Done()
+			// avoid explosion of tasks with name f.report().Info.Name
+			ctx, endTask := trace.WithTaskAndSpan(ctx, "repl-fs", f.report().Info.Name)
+			defer endTask()
+			f.do(ctx, stepQueue, prevs[f])
+			f.l.HoldWhile(func() {
+				// every return from f means it's unblocked...
+				f.blockedOn = report.FsBlockedOnNothing
+			})
+		}(f)
+	}
+	a.l.DropWhile(func() {
+		fssesDone.Wait()
+	})
+	a.finishedAt = time.Now()
+}
+
+func (f *fs) debug(format string, args ...interface{}) {
+	debugPrefix("fs=%s", f.fs.ReportInfo().Name)(format, args...)
+}
+
+// wake up children that watch for f.{planning.{err,done},planned.{step,stepErr}}
+func (f *fs) initialRepOrdWakeupChildren() {
+	var children []string
+	for _, c := range f.initialRepOrd.children {
+		// no locking required, c.fs does not change
+		children = append(children, c.fs.ReportInfo().Name)
+	}
+	f.debug("wakeup children %s", children)
+	for _, child := range f.initialRepOrd.children {
+		select {
+		// no locking required, child.initialRepOrd does not change
+		case child.initialRepOrd.parentDidUpdate <- struct{}{}:
+		default:
+		}
+	}
+}
+
+func (f *fs) do(ctx context.Context, pq *stepQueue, prev *fs) {
+
+	defer f.l.Lock().Unlock()
+	defer f.initialRepOrdWakeupChildren()
+
+	// get planned steps from replication logic
+	var psteps []Step
+	var errTime time.Time
+	var err error
+	f.blockedOn = report.FsBlockedOnPlanningStepQueue
+	f.l.DropWhile(func() {
+		// TODO hacky
+		// choose target time that is earlier than any snapshot, so fs planning is always prioritized
+		targetDate := time.Unix(0, 0)
+		defer pq.WaitReady(ctx, f, targetDate)()
+		f.l.HoldWhile(func() {
+			// transition before we call PlanFS
+			f.blockedOn = report.FsBlockedOnNothing
+		})
+		psteps, err = f.fs.PlanFS(ctx) // no shadow
+		errTime = time.Now()           // no shadow
+	})
+	if err != nil {
+		f.planning.err = newTimedError(err, errTime)
+		return
+	}
+	for _, pstep := range psteps {
+		step := &step{
+			l:    f.l,
+			step: pstep,
+		}
+		f.planned.steps = append(f.planned.steps, step)
+	}
+	// we're not done planning yet, f.planned.steps might still be changed by next block
+	// => don't set f.planning.done just yet
+	f.debug("initial len(fs.planned.steps) = %d", len(f.planned.steps))
+
+	// for not-first attempts that succeeded in planning, only allow fs.planned.steps
+	// up to and including the originally planned target snapshot
+	if prev != nil && prev.planning.done && prev.planning.err == nil {
+		f.debug("attempting to correlate plan with previous attempt to find out what is left to do")
+		// find the highest of the previously uncompleted steps for which we can also find a step
+		// in our current plan
+		prevUncompleted := prev.planned.steps[prev.planned.step:]
+		if len(prevUncompleted) == 0 || len(f.planned.steps) == 0 {
+			f.debug("no steps planned in previous attempt or this attempt, no correlation necessary len(prevUncompleted)=%d len(f.planned.steps)=%d", len(prevUncompleted), len(f.planned.steps))
+		} else {
+			var target struct{ prev, cur int }
+			target.prev = -1
+			target.cur = -1
+		out:
+			for p := len(prevUncompleted) - 1; p >= 0; p-- {
+				for q := len(f.planned.steps) - 1; q >= 0; q-- {
+					if prevUncompleted[p].step.TargetEquals(f.planned.steps[q].step) {
+						target.prev = p
+						target.cur = q
+						break out
+					}
+				}
+			}
+			if target.prev == -1 || target.cur == -1 {
+				f.debug("no correlation possible between previous attempt and this attempt's plan")
+				f.planning.err = newTimedError(fmt.Errorf("cannot correlate previously failed attempt to current plan"), time.Now())
+				return
+			}
+
+			f.planned.steps = f.planned.steps[0:target.cur]
+			f.debug("found correlation, new steps are len(fs.planned.steps) = %d", len(f.planned.steps))
+		}
+	} else {
+		f.debug("previous attempt does not exist or did not finish planning, no correlation possible, taking this attempt's plan as is")
+	}
+
+	// now we are done planning (f.planned.steps won't change from now on)
+	f.planning.done = true
+
+	// wait for parents' initial replication
+	f.blockedOn = report.FsBlockedOnParentInitialRepl
+	var parents []string
+	for _, p := range f.initialRepOrd.parents {
+		parents = append(parents, p.fs.ReportInfo().Name)
+	}
+	f.debug("wait for parents %s", parents)
+	for {
+		var initialReplicatingParentsWithErrors []string
+		allParentsPresentOnReceiver := true
+		f.l.DropWhile(func() {
+			for _, p := range f.initialRepOrd.parents {
+				p.l.HoldWhile(func() {
+					// (get the preconditions that allow us to inspect p.planned)
+					parentHasPlanningDone := p.planning.done && p.planning.err == nil
+					if !parentHasPlanningDone {
+						// if the parent couldn't be planned, we cannot know whether it needs initial replication
+						// or incremental replication => be conservative and assume it was initial replication
+						allParentsPresentOnReceiver = false
+						if p.planning.err != nil {
+							initialReplicatingParentsWithErrors = append(initialReplicatingParentsWithErrors, p.fs.ReportInfo().Name)
+						}
+						return
+					}
+					// now allowed to inspect p.planned
+
+					// if there are no steps to be done, the filesystem must exist on the receiving side
+					// (otherwise we'd replicate it, and there would be a step for that)
+					// (FIXME hardcoded initial replication policy, assuming the policy will always do _some_ initial replication)
+					parentHasNoSteps := len(p.planned.steps) == 0
+
+					// OR if it has completed at least one step
+					// (remember that .step points to the next step to be done)
+					// (TODO technically, we could make this step ready in the moment the recv-side
+					//  dataset exists, i.e. after the first few megabytes of transferred data, but we'd have to ask the receiver for that -> poll ListFilesystems RPC)
+					parentHasTakenAtLeastOneSuccessfulStep := !parentHasNoSteps && p.planned.step >= 1
+
+					parentFirstStepIsIncremental := // no need to lock for .report() because step.l == it's fs.l
+						len(p.planned.steps) > 0 && p.planned.steps[0].report().IsIncremental()
+
+					f.debug("parentHasNoSteps=%v parentFirstStepIsIncremental=%v parentHasTakenAtLeastOneSuccessfulStep=%v",
+						parentHasNoSteps, parentFirstStepIsIncremental, parentHasTakenAtLeastOneSuccessfulStep)
+
+					// If the parent is a placeholder on the sender, `parentHasNoSteps` is true because we plan no steps for sender placeholders.
+					// The receiver will create the necessary placeholders when they start receiving the first non-placeholder child filesystem.
+					parentPresentOnReceiver := parentHasNoSteps || parentFirstStepIsIncremental || parentHasTakenAtLeastOneSuccessfulStep
+
+					allParentsPresentOnReceiver = allParentsPresentOnReceiver && parentPresentOnReceiver // no shadow
+
+					if !parentPresentOnReceiver && p.planned.stepErr != nil {
+						initialReplicatingParentsWithErrors = append(initialReplicatingParentsWithErrors, p.fs.ReportInfo().Name)
+					}
+
+				})
+			}
+		})
+
+		if len(initialReplicatingParentsWithErrors) > 0 {
+			f.planned.stepErr = newTimedError(fmt.Errorf("parent(s) failed during initial replication: %s", initialReplicatingParentsWithErrors), time.Now())
+			return
+		}
+
+		if allParentsPresentOnReceiver {
+			break // good to go
+		}
+
+		// wait for wakeups from parents, then check again
+		// lock must not be held while waiting in order for reporting to work
+		f.l.DropWhile(func() {
+			select {
+			case <-ctx.Done():
+				f.planned.stepErr = newTimedError(ctx.Err(), time.Now())
+			case <-f.initialRepOrd.parentDidUpdate:
+				// loop
+			}
+		})
+		if f.planned.stepErr != nil {
+			return
+		}
+	}
+
+	f.debug("all parents ready, start replication %s", parents)
+
+	// do our steps
+	for i, s := range f.planned.steps {
+		// lock must not be held while executing step in order for reporting to work
+		f.l.DropWhile(func() {
+			// wait for parallel replication
+			targetDate := s.step.TargetDate()
+			f.l.HoldWhile(func() { f.blockedOn = report.FsBlockedOnReplStepQueue })
+			defer pq.WaitReady(ctx, f, targetDate)()
+			f.l.HoldWhile(func() { f.blockedOn = report.FsBlockedOnNothing })
+			// do the step
+			ctx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("%#v", s.step.ReportInfo()))
+			defer endSpan()
+			err, errTime = s.step.Step(ctx), time.Now() // no shadow
+		})
+
+		if err != nil {
+			f.planned.stepErr = newTimedError(err, errTime)
+			break
+		}
+		f.planned.step = i + 1 // fs.planned.step must be == len(fs.planned.steps) if all went OK
+
+		f.initialRepOrdWakeupChildren()
+	}
+
+}
+
+// caller must hold lock l
+func (r *run) report() *report.Report {
+	report := &report.Report{
+		Attempts:           make([]*report.AttemptReport, len(r.attempts)),
+		StartAt:            r.startedAt,
+		FinishAt:           r.finishedAt,
+		WaitReconnectSince: r.waitReconnect.begin,
+		WaitReconnectUntil: r.waitReconnect.end,
+		WaitReconnectError: r.waitReconnectError.IntoReportError(),
+	}
+	for i := range report.Attempts {
+		report.Attempts[i] = r.attempts[i].report()
+	}
+	return report
+}
+
+// caller must hold lock l
+func (a *attempt) report() *report.AttemptReport {
+
+	r := &report.AttemptReport{
+		// State is set below
+		Filesystems: make([]*report.FilesystemReport, len(a.fss)),
+		StartAt:     a.startedAt,
+		FinishAt:    a.finishedAt,
+		PlanError:   a.planErr.IntoReportError(),
+	}
+
+	for i := range r.Filesystems {
+		r.Filesystems[i] = a.fss[i].report()
+	}
+
+	var state report.AttemptState
+	if a.planErr == nil && a.fss == nil {
+		state = report.AttemptPlanning
+	} else if a.planErr != nil && a.fss == nil {
+		state = report.AttemptPlanningError
+	} else if a.planErr == nil && a.fss != nil {
+		if a.finishedAt.IsZero() {
+			state = report.AttemptFanOutFSs
+		} else {
+			fsWithError := false
+			for _, s := range r.Filesystems {
+				fsWithError = fsWithError || s.Error() != nil
+			}
+			state = report.AttemptDone
+			if fsWithError {
+				state = report.AttemptFanOutError
+			}
+		}
+	} else {
+		panic(fmt.Sprintf("attempt.planErr and attempt.fss must not both be != nil:\n%s\n%s", pretty.Sprint(a.planErr), pretty.Sprint(a.fss)))
+	}
+	r.State = state
+
+	return r
+}
+
+// caller must hold lock l
+func (f *fs) report() *report.FilesystemReport {
+	state := report.FilesystemPlanningErrored
+	if f.planning.err == nil {
+		if f.planning.done {
+			if f.planned.stepErr != nil {
+				state = report.FilesystemSteppingErrored
+			} else if f.planned.step < len(f.planned.steps) {
+				state = report.FilesystemStepping
+			} else {
+				state = report.FilesystemDone
+			}
+		} else {
+			state = report.FilesystemPlanning
+		}
+	}
+	r := &report.FilesystemReport{
+		Info:        f.fs.ReportInfo(),
+		State:       state,
+		BlockedOn:   f.blockedOn,
+		PlanError:   f.planning.err.IntoReportError(),
+		StepError:   f.planned.stepErr.IntoReportError(),
+		Steps:       make([]*report.StepReport, len(f.planned.steps)),
+		CurrentStep: f.planned.step,
+	}
+	for i := range r.Steps {
+		r.Steps[i] = f.planned.steps[i].report()
+	}
+	return r
+}
+
+// caller must hold lock l
+func (s *step) report() *report.StepReport {
+	r := &report.StepReport{
+		Info: s.step.ReportInfo(),
+	}
+	return r
+}
+
+//go:generate enumer -type=errorClass
+type errorClass int
+
+const (
+	errorClassPermanent errorClass = iota
+	errorClassTemporaryConnectivityRelated
+)
+
+type errorReport struct {
+	flattened []*timedError
+	// sorted DESCending by err time
+	byClass map[errorClass][]*timedError
+}
+
+// caller must hold lock l
+func (a *attempt) errorReport() *errorReport {
+	r := &errorReport{}
+	if a.planErr != nil {
+		r.flattened = append(r.flattened, a.planErr)
+	}
+	for _, fs := range a.fss {
+		if fs.planning.done && fs.planning.err != nil {
+			r.flattened = append(r.flattened, fs.planning.err)
+		} else if fs.planning.done && fs.planned.stepErr != nil {
+			r.flattened = append(r.flattened, fs.planned.stepErr)
+		}
+	}
+
+	// build byClass
+	{
+		r.byClass = make(map[errorClass][]*timedError)
+		putClass := func(err *timedError, class errorClass) {
+			errs := r.byClass[class]
+			errs = append(errs, err)
+			r.byClass[class] = errs
+		}
+		for _, err := range r.flattened {
+			if neterr, ok := err.Err.(net.Error); ok && neterr.Timeout() {
+				putClass(err, errorClassTemporaryConnectivityRelated)
+				continue
+			}
+			if st, ok := status.FromError(err.Err); ok && st.Code() == codes.Unavailable {
+				// technically, codes.Unavailable could be returned by the gRPC endpoint, indicating overload, etc.
+				// for now, let's assume it only happens for connectivity issues, as specified in
+				// https://grpc.io/grpc/core/md_doc_statuscodes.html
+				putClass(err, errorClassTemporaryConnectivityRelated)
+				continue
+			}
+			putClass(err, errorClassPermanent)
+		}
+		for _, errs := range r.byClass {
+			sort.Slice(errs, func(i, j int) bool {
+				return errs[i].Time.After(errs[j].Time) // sort descendingly
+			})
+		}
+	}
+
+	return r
+}
+
+func (r *errorReport) AnyError() *timedError {
+	for _, err := range r.flattened {
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *errorReport) MostRecent() (err *timedError, errClass errorClass) {
+	for class, errs := range r.byClass {
+		// errs are sorted descendingly during construction
+		if len(errs) > 0 && (err == nil || errs[0].Time.After(err.Time)) {
+			err = errs[0]
+			errClass = class
+		}
+	}
+	return
+}

internal/replication/driver/replication_driver_debug.go

@@ -0,0 +1,31 @@
+package driver
+
+import (
+	"fmt"
+	"os"
+)
+
+var debugEnabled bool = false
+
+func init() {
+	if os.Getenv("ZREPL_REPLICATION_DRIVER_DEBUG") != "" {
+		debugEnabled = true
+	}
+}
+
+//nolint:deadcode,unused
+func debug(format string, args ...interface{}) {
+	if debugEnabled {
+		fmt.Fprintf(os.Stderr, "repl: driver: %s\n", fmt.Sprintf(format, args...))
+	}
+}
+
+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{}) {
+		debug("%s: %s", prefix, fmt.Sprintf(format, args...))
+	}
+}

internal/replication/driver/replication_driver_logging.go

@@ -0,0 +1,12 @@
+package driver
+
+import (
+	"context"
+
+	"github.com/zrepl/zrepl/internal/daemon/logging"
+	"github.com/zrepl/zrepl/internal/logger"
+)
+
+func getLog(ctx context.Context) logger.Logger {
+	return logging.GetLogger(ctx, logging.SubsysReplication)
+}

internal/replication/driver/replication_driver_test.go

@@ -0,0 +1,224 @@
+package driver
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sort"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/zrepl/zrepl/internal/daemon/logging/trace"
+
+	"github.com/zrepl/zrepl/internal/replication/report"
+
+	"github.com/stretchr/testify/assert"
+
+	jsondiff "github.com/yudai/gojsondiff"
+	jsondiffformatter "github.com/yudai/gojsondiff/formatter"
+)
+
+type mockPlanner struct {
+	stepCounter uint32
+	fss         []FS // *mockFS
+}
+
+func (p *mockPlanner) Plan(ctx context.Context) ([]FS, error) {
+	time.Sleep(1 * time.Second)
+	p.fss = []FS{
+		&mockFS{
+			&p.stepCounter,
+			"zroot/one",
+			nil,
+		},
+		&mockFS{
+			&p.stepCounter,
+			"zroot/two",
+			nil,
+		},
+	}
+	return p.fss, nil
+}
+
+func (p *mockPlanner) WaitForConnectivity(context.Context) error {
+	return nil
+}
+
+type mockFS struct {
+	globalStepCounter *uint32
+	name              string
+	steps             []Step
+}
+
+func (f *mockFS) EqualToPreviousAttempt(other FS) bool {
+	return f.name == other.(*mockFS).name
+}
+
+func (f *mockFS) PlanFS(ctx context.Context) ([]Step, error) {
+	if f.steps != nil {
+		panic("PlanFS used twice")
+	}
+	switch f.name {
+	case "zroot/one":
+		f.steps = []Step{
+			&mockStep{
+				fs:         f,
+				ident:      "a",
+				duration:   1 * time.Second,
+				targetDate: time.Unix(2, 0),
+			},
+			&mockStep{
+				fs:         f,
+				ident:      "b",
+				duration:   1 * time.Second,
+				targetDate: time.Unix(10, 0),
+			},
+			&mockStep{
+				fs:         f,
+				ident:      "c",
+				duration:   1 * time.Second,
+				targetDate: time.Unix(20, 0),
+			},
+		}
+	case "zroot/two":
+		f.steps = []Step{
+			&mockStep{
+				fs:         f,
+				ident:      "u",
+				duration:   500 * time.Millisecond,
+				targetDate: time.Unix(15, 0),
+			},
+			&mockStep{
+				fs:         f,
+				duration:   500 * time.Millisecond,
+				ident:      "v",
+				targetDate: time.Unix(30, 0),
+			},
+		}
+	default:
+		panic("unimplemented")
+	}
+
+	return f.steps, nil
+}
+
+func (f *mockFS) ReportInfo() *report.FilesystemInfo {
+	return &report.FilesystemInfo{Name: f.name}
+}
+
+type mockStep struct {
+	fs         *mockFS
+	ident      string
+	duration   time.Duration
+	targetDate time.Time
+
+	// filled by method Step
+	globalCtr uint32
+}
+
+func (f *mockStep) String() string {
+	return fmt.Sprintf("%s{%s} targetDate=%s globalCtr=%v", f.fs.name, f.ident, f.targetDate, f.globalCtr)
+}
+
+func (f *mockStep) Step(ctx context.Context) error {
+	f.globalCtr = atomic.AddUint32(f.fs.globalStepCounter, 1)
+	time.Sleep(f.duration)
+	return nil
+}
+
+func (f *mockStep) TargetEquals(s Step) bool {
+	return f.ident == s.(*mockStep).ident
+}
+
+func (f *mockStep) TargetDate() time.Time {
+	return f.targetDate
+}
+
+func (f *mockStep) ReportInfo() *report.StepInfo {
+	return &report.StepInfo{From: f.ident, To: f.ident, BytesExpected: 100, BytesReplicated: 25}
+}
+
+// TODO: add meaningful validation (i.e. actual checks)
+// Since the stepqueue is not deterministic due to scheduler jitter,
+// we cannot test for any definitive sequence of steps here.
+// Such checks would further only be sensible for a non-concurrent step-queue,
+// but we're going to have concurrent replication in the future.
+//
+// For the time being, let's just exercise the code a bit.
+func TestReplication(t *testing.T) {
+
+	ctx := context.Background()
+	defer trace.WithTaskFromStackUpdateCtx(&ctx)()
+
+	mp := &mockPlanner{}
+	driverConfig := Config{
+		StepQueueConcurrency:     1,
+		MaxAttempts:              1,
+		ReconnectHardFailTimeout: 1 * time.Second,
+	}
+	getReport, wait := Do(ctx, driverConfig, mp)
+	begin := time.Now()
+	fireAt := []time.Duration{
+		// the following values are relative to the start
+		500 * time.Millisecond,  // planning
+		1500 * time.Millisecond, // nothing is done, a is running
+		2500 * time.Millisecond, // a done,  b running
+		3250 * time.Millisecond, // a,b done, u running
+		3750 * time.Millisecond, // a,b,u done, c running
+		4750 * time.Millisecond, // a,b,u,c done, v running
+		5250 * time.Millisecond, // a,b,u,c,v done
+	}
+	reports := make([]*report.Report, len(fireAt))
+	for i := range fireAt {
+		sleepUntil := begin.Add(fireAt[i])
+		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.Since(waitBegin)
+	assert.True(t, waitDuration < 10*time.Millisecond, "%v", waitDuration) // and that's gracious
+
+	prev, err := json.Marshal(reports[0])
+	require.NoError(t, err)
+	for _, r := range reports[1:] {
+		this, err := json.Marshal(r)
+		require.NoError(t, err)
+		differ := jsondiff.New()
+		diff, err := differ.Compare(prev, this)
+		require.NoError(t, err)
+		df := jsondiffformatter.NewDeltaFormatter()
+		_, err = df.Format(diff)
+		require.NoError(t, err)
+		// uncomment the following line to get json diffs between each captured step
+		// t.Logf("%s", res)
+		prev, err = json.Marshal(r)
+		require.NoError(t, err)
+	}
+
+	steps := make([]*mockStep, 0)
+	for _, fs := range mp.fss {
+		for _, step := range fs.(*mockFS).steps {
+			steps = append(steps, step.(*mockStep))
+		}
+	}
+
+	// sort steps in pq order (although, remember, pq is not deterministic)
+	sort.Slice(steps, func(i, j int) bool {
+		return steps[i].targetDate.Before(steps[j].targetDate)
+	})
+
+	// manual inspection of the globalCtr value should show that, despite
+	// scheduler-dependent behavior of pq, steps should generally be taken
+	// from oldest to newest target date (globally, not per FS).
+	t.Logf("steps sorted by target date:")
+	for _, step := range steps {
+		t.Logf("\t%s", step)
+	}
+
+}

internal/replication/driver/replication_stepqueue.go

@@ -0,0 +1,166 @@
+package driver
+
+import (
+	"container/heap"
+	"context"
+	"time"
+
+	"github.com/zrepl/zrepl/internal/daemon/logging/trace"
+	"github.com/zrepl/zrepl/internal/util/chainlock"
+)
+
+type stepQueueRec struct {
+	ident      interface{}
+	targetDate time.Time
+	wakeup     chan StepCompletedFunc
+}
+
+type stepQueue struct {
+	stop chan struct{}
+	reqs chan stepQueueRec
+}
+
+type stepQueueHeapItem struct {
+	idx int
+	req stepQueueRec
+}
+type stepQueueHeap []*stepQueueHeapItem
+
+func (h stepQueueHeap) Less(i, j int) bool {
+	return h[i].req.targetDate.Before(h[j].req.targetDate)
+}
+
+func (h stepQueueHeap) Swap(i, j int) {
+	h[i], h[j] = h[j], h[i]
+	h[i].idx = i
+	h[j].idx = j
+}
+
+func (h stepQueueHeap) Len() int {
+	return len(h)
+}
+
+func (h *stepQueueHeap) Push(elem interface{}) {
+	hitem := elem.(*stepQueueHeapItem)
+	hitem.idx = h.Len()
+	*h = append(*h, hitem)
+}
+
+func (h *stepQueueHeap) Pop() interface{} {
+	elem := (*h)[h.Len()-1]
+	elem.idx = -1
+	*h = (*h)[:h.Len()-1]
+	return elem
+}
+
+// returned stepQueue must be closed with method Close
+func newStepQueue() *stepQueue {
+	q := &stepQueue{
+		stop: make(chan struct{}),
+		reqs: make(chan stepQueueRec),
+	}
+	return q
+}
+
+// the returned done function must be called to free resources
+// allocated by the call to Start
+//
+// No WaitReady calls must be active at the time done is called
+// The behavior of calling WaitReady after done was called is undefined
+func (q *stepQueue) Start(concurrency int) (done func()) {
+	if concurrency < 1 {
+		panic("concurrency must be >= 1")
+	}
+	// l protects pending and queueItems
+	l := chainlock.New()
+	pendingCond := l.NewCond()
+	// priority queue
+	pending := &stepQueueHeap{}
+	// ident => queueItem
+	queueItems := make(map[interface{}]*stepQueueHeapItem)
+	// stopped is used for cancellation of "wake" goroutine
+	stopped := false
+	active := 0
+	go func() { // "stopper" goroutine
+		<-q.stop
+		defer l.Lock().Unlock()
+		stopped = true
+		pendingCond.Broadcast()
+	}()
+	go func() { // "reqs" goroutine
+		for {
+			select {
+			case <-q.stop:
+				select {
+				case <-q.reqs:
+					panic("WaitReady call active while calling Close")
+				default:
+					return
+				}
+			case req := <-q.reqs:
+				func() {
+					defer l.Lock().Unlock()
+					if _, ok := queueItems[req.ident]; ok {
+						panic("WaitReady must not be called twice for the same ident")
+					}
+					qitem := &stepQueueHeapItem{
+						req: req,
+					}
+					queueItems[req.ident] = qitem
+					heap.Push(pending, qitem)
+					pendingCond.Broadcast()
+				}()
+			}
+		}
+	}()
+	go func() { // "wake" goroutine
+		defer l.Lock().Unlock()
+		for {
+
+			for !stopped && (active >= concurrency || pending.Len() == 0) {
+				pendingCond.Wait()
+			}
+			if stopped {
+				return
+			}
+			if pending.Len() <= 0 {
+				return
+			}
+			active++
+			next := heap.Pop(pending).(*stepQueueHeapItem).req
+			delete(queueItems, next.ident)
+
+			next.wakeup <- func() {
+				defer l.Lock().Unlock()
+				active--
+				pendingCond.Broadcast()
+			}
+		}
+	}()
+
+	done = func() {
+		close(q.stop)
+	}
+	return done
+}
+
+type StepCompletedFunc func()
+
+func (q *stepQueue) sendAndWaitForWakeup(ident interface{}, targetDate time.Time) StepCompletedFunc {
+	req := stepQueueRec{
+		ident,
+		targetDate,
+		make(chan StepCompletedFunc),
+	}
+	q.reqs <- req
+	return <-req.wakeup
+}
+
+// Wait for the ident with targetDate to be selected to run.
+func (q *stepQueue) WaitReady(ctx context.Context, ident interface{}, targetDate time.Time) StepCompletedFunc {
+	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
+	if targetDate.IsZero() {
+		panic("targetDate of zero is reserved for marking Done")
+	}
+	return q.sendAndWaitForWakeup(ident, targetDate)
+}

internal/replication/driver/replication_stepqueue_test.go

@@ -0,0 +1,197 @@
+package driver
+
+import (
+	"context"
+	"fmt"
+	"math"
+	"sort"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/montanaflynn/stats"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/zrepl/zrepl/internal/daemon/logging/trace"
+	"github.com/zrepl/zrepl/internal/util/zreplcircleci"
+)
+
+func TestPqNotconcurrent(t *testing.T) {
+	zreplcircleci.SkipOnCircleCI(t, "because it relies on scheduler responsiveness < 500ms")
+
+	ctx, end := trace.WithTaskFromStack(context.Background())
+	defer end()
+	var ctr uint32
+	q := newStepQueue()
+	var wg sync.WaitGroup
+	wg.Add(4)
+	go func() {
+		ctx, end := trace.WithTaskFromStack(ctx)
+		defer end()
+		defer wg.Done()
+		defer q.WaitReady(ctx, "1", time.Unix(9999, 0))()
+		ret := atomic.AddUint32(&ctr, 1)
+		assert.Equal(t, uint32(1), ret)
+		time.Sleep(1 * time.Second)
+	}()
+
+	// give goroutine "1" 500ms to enter queue, get the active slot and enter time.Sleep
+	defer q.Start(1)()
+	time.Sleep(500 * time.Millisecond)
+
+	// while "1" is still running, queue in "2", "3" and "4"
+	go func() {
+		ctx, end := trace.WithTaskFromStack(ctx)
+		defer end()
+		defer wg.Done()
+		defer q.WaitReady(ctx, "2", time.Unix(2, 0))()
+		ret := atomic.AddUint32(&ctr, 1)
+		assert.Equal(t, uint32(2), ret)
+	}()
+	go func() {
+		ctx, end := trace.WithTaskFromStack(ctx)
+		defer end()
+		defer wg.Done()
+		defer q.WaitReady(ctx, "3", time.Unix(3, 0))()
+		ret := atomic.AddUint32(&ctr, 1)
+		assert.Equal(t, uint32(3), ret)
+	}()
+	go func() {
+		ctx, end := trace.WithTaskFromStack(ctx)
+		defer end()
+		defer wg.Done()
+		defer q.WaitReady(ctx, "4", time.Unix(4, 0))()
+		ret := atomic.AddUint32(&ctr, 1)
+		assert.Equal(t, uint32(4), ret)
+	}()
+
+	wg.Wait()
+}
+
+type record struct {
+	fs        int
+	step      int
+	globalCtr uint32
+	wakeAt    time.Duration // relative to begin
+}
+
+func (r record) String() string {
+	return fmt.Sprintf("fs %08d step %08d globalCtr %08d wakeAt %2.8f", r.fs, r.step, r.globalCtr, r.wakeAt.Seconds())
+}
+
+// This tests uses stepPq concurrently, simulating the following scenario:
+// Given a number of filesystems F, each filesystem has N steps to take.
+// The number of concurrent steps is limited to C.
+// The target date for each step is the step number N.
+// Hence, there are always F filesystems runnable (calling WaitReady)
+// The priority queue prioritizes steps with lower target data (= lower step number).
+// Hence, all steps with lower numbers should be woken up before steps with higher numbers.
+// However, scheduling is not 100% deterministic (runtime, OS scheduler, etc).
+// Hence, perform some statistics on the wakeup times and assert that the mean wakeup
+// times for each step are close together.
+func TestPqConcurrent(t *testing.T) {
+	zreplcircleci.SkipOnCircleCI(t, "because it relies on scheduler responsiveness < 500ms")
+
+	ctx, end := trace.WithTaskFromStack(context.Background())
+	defer end()
+
+	q := newStepQueue()
+	var wg sync.WaitGroup
+	filesystems := 100
+	stepsPerFS := 20
+	sleepTimePerStep := 50 * time.Millisecond
+	wg.Add(filesystems)
+	var globalCtr uint32
+
+	begin := time.Now()
+	records := make(chan []record, filesystems)
+	for fs := 0; fs < filesystems; fs++ {
+		go func(fs int) {
+			ctx, end := trace.WithTaskFromStack(ctx)
+			defer end()
+			defer wg.Done()
+			recs := make([]record, 0)
+			for step := 0; step < stepsPerFS; step++ {
+				pos := atomic.AddUint32(&globalCtr, 1)
+				t := time.Unix(int64(step), 0)
+				done := q.WaitReady(ctx, fs, t)
+				wakeAt := time.Since(begin)
+				time.Sleep(sleepTimePerStep)
+				done()
+				recs = append(recs, record{fs, step, pos, wakeAt})
+			}
+			records <- recs
+		}(fs)
+	}
+	concurrency := 5
+	defer q.Start(concurrency)()
+	wg.Wait()
+	close(records)
+	t.Logf("loop done")
+
+	flattenedRecs := make([]record, 0)
+	for recs := range records {
+		flattenedRecs = append(flattenedRecs, recs...)
+	}
+
+	sort.Slice(flattenedRecs, func(i, j int) bool {
+		return flattenedRecs[i].globalCtr < flattenedRecs[j].globalCtr
+	})
+
+	wakeTimesByStep := map[int][]float64{}
+	for _, rec := range flattenedRecs {
+		wakeTimes, ok := wakeTimesByStep[rec.step]
+		if !ok {
+			wakeTimes = []float64{}
+		}
+		wakeTimes = append(wakeTimes, rec.wakeAt.Seconds())
+		wakeTimesByStep[rec.step] = wakeTimes
+	}
+
+	meansByStepId := make([]float64, stepsPerFS)
+	interQuartileRangesByStepIdx := make([]float64, stepsPerFS)
+	for step := 0; step < stepsPerFS; step++ {
+		t.Logf("step %d", step)
+		mean, _ := stats.Mean(wakeTimesByStep[step])
+		meansByStepId[step] = mean
+		t.Logf("\tmean: %v", mean)
+		median, _ := stats.Median(wakeTimesByStep[step])
+		t.Logf("\tmedian: %v", median)
+		midhinge, _ := stats.Midhinge(wakeTimesByStep[step])
+		t.Logf("\tmidhinge: %v", midhinge)
+		min, _ := stats.Min(wakeTimesByStep[step])
+		t.Logf("\tmin: %v", min)
+		max, _ := stats.Max(wakeTimesByStep[step])
+		t.Logf("\tmax: %v", max)
+		quartiles, _ := stats.Quartile(wakeTimesByStep[step])
+		t.Logf("\t%#v", quartiles)
+		interQuartileRange, _ := stats.InterQuartileRange(wakeTimesByStep[step])
+		t.Logf("\tinter-quartile range: %v", interQuartileRange)
+		interQuartileRangesByStepIdx[step] = interQuartileRange
+	}
+
+	iqrMean, _ := stats.Mean(interQuartileRangesByStepIdx)
+	t.Logf("inter-quartile-range mean: %v", iqrMean)
+	iqrDev, _ := stats.StandardDeviation(interQuartileRangesByStepIdx)
+	t.Logf("inter-quartile-range deviation: %v", iqrDev)
+
+	// each step should have the same "distribution" (=~ "spread")
+	assert.True(t, iqrDev < 0.01)
+
+	minTimeForAllStepsWithIdxI := sleepTimePerStep.Seconds() * float64(filesystems) / float64(concurrency)
+	t.Logf("minTimeForAllStepsWithIdxI = %11.8f", minTimeForAllStepsWithIdxI)
+	for i, mean := range meansByStepId {
+		// we can't just do (i + 0.5) * minTimeforAllStepsWithIdxI
+		// because this doesn't account for drift
+		idealMean := 0.5 * minTimeForAllStepsWithIdxI
+		if i > 0 {
+			previousMean := meansByStepId[i-1]
+			idealMean = previousMean + minTimeForAllStepsWithIdxI
+		}
+		deltaFromIdeal := idealMean - mean
+		t.Logf("step %02d delta from ideal mean wake time: %11.8f - %11.8f = %11.8f", i, idealMean, mean, deltaFromIdeal)
+		assert.True(t, math.Abs(deltaFromIdeal) < 0.05)
+	}
+
+}

internal/replication/logic/diff/diff.go

@@ -0,0 +1,174 @@
+package diff
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+
+	. "github.com/zrepl/zrepl/internal/replication/logic/pdu"
+)
+
+type ConflictNoCommonAncestor struct {
+	SortedSenderVersions, SortedReceiverVersions []*FilesystemVersion
+}
+
+func (c *ConflictNoCommonAncestor) Error() string {
+	var buf strings.Builder
+	buf.WriteString("no common snapshot or suitable bookmark between sender and receiver")
+	if len(c.SortedReceiverVersions) > 0 || len(c.SortedSenderVersions) > 0 {
+		buf.WriteString(":\n  sorted sender versions:\n")
+		for _, v := range c.SortedSenderVersions {
+			fmt.Fprintf(&buf, "    %s\n", v.RelName())
+		}
+		buf.WriteString("  sorted receiver versions:\n")
+		for _, v := range c.SortedReceiverVersions {
+			fmt.Fprintf(&buf, "    %s\n", v.RelName())
+		}
+	}
+	return buf.String()
+}
+
+type ConflictDiverged struct {
+	SortedSenderVersions, SortedReceiverVersions []*FilesystemVersion
+	CommonAncestor                               *FilesystemVersion
+	SenderOnly, ReceiverOnly                     []*FilesystemVersion
+}
+
+func (c *ConflictDiverged) Error() string {
+	var buf strings.Builder
+	buf.WriteString("the receiver's latest snapshot is not present on sender:\n")
+	fmt.Fprintf(&buf, "  last common: %s\n", c.CommonAncestor.RelName())
+	fmt.Fprintf(&buf, "  sender-only:\n")
+	for _, v := range c.SenderOnly {
+		fmt.Fprintf(&buf, "    %s\n", v.RelName())
+	}
+	fmt.Fprintf(&buf, "  receiver-only:\n")
+	for _, v := range c.ReceiverOnly {
+		fmt.Fprintf(&buf, "    %s\n", v.RelName())
+	}
+	return buf.String()
+}
+
+type ConflictNoSenderSnapshots struct{}
+
+func (c *ConflictNoSenderSnapshots) Error() string {
+	return "no snapshots available on sender side"
+}
+
+type ConflictMostRecentSnapshotAlreadyPresent struct {
+	SortedSenderVersions, SortedReceiverVersions []*FilesystemVersion
+	CommonAncestor                               *FilesystemVersion
+}
+
+func (c *ConflictMostRecentSnapshotAlreadyPresent) Error() string {
+	var buf strings.Builder
+	fmt.Fprintf(&buf, "the most recent sender snapshot is already present on the receiver (guid=%v, name=%q)", c.CommonAncestor.GetGuid(), c.CommonAncestor.RelName())
+	return buf.String()
+}
+
+func SortVersionListByCreateTXGThenBookmarkLTSnapshot(fsvslice []*FilesystemVersion) []*FilesystemVersion {
+	lesser := func(s []*FilesystemVersion) func(i, j int) bool {
+		return func(i, j int) bool {
+			if s[i].CreateTXG < s[j].CreateTXG {
+				return true
+			}
+			if s[i].CreateTXG == s[j].CreateTXG {
+				//  Bookmark < Snapshot
+				return s[i].Type == FilesystemVersion_Bookmark && s[j].Type == FilesystemVersion_Snapshot
+			}
+			return false
+		}
+	}
+	if sort.SliceIsSorted(fsvslice, lesser(fsvslice)) {
+		return fsvslice
+	}
+	sorted := make([]*FilesystemVersion, len(fsvslice))
+	copy(sorted, fsvslice)
+	sort.Slice(sorted, lesser(sorted))
+	return sorted
+}
+
+func StripBookmarksFromVersionList(fsvslice []*FilesystemVersion) []*FilesystemVersion {
+	fslice := make([]*FilesystemVersion, 0, len(fsvslice))
+	for _, fv := range fsvslice {
+		if fv.Type != FilesystemVersion_Bookmark {
+			fslice = append(fslice, fv)
+		}
+	}
+	return fslice
+}
+
+func IncrementalPath(receiver, sender []*FilesystemVersion) (incPath []*FilesystemVersion, conflict error) {
+
+	// Receive-side bookmarks can't be used as incremental-from,
+	// and don't cause recv to fail if there is a newer bookmark than incremetal-form on the receiver.
+	// So, simply mask them out.
+	// This will also hide them in the report, but it keeps the code in this function simple,
+	// and a user who complains about them missing in a conflict message will likely require
+	// more education about bookmarks than a slightly more accurate error message. They'll get
+	// that when they open an issue.
+	receiver = StripBookmarksFromVersionList(receiver)
+
+	receiver = SortVersionListByCreateTXGThenBookmarkLTSnapshot(receiver)
+	sender = SortVersionListByCreateTXGThenBookmarkLTSnapshot(sender)
+
+	var mrcaCandidate struct {
+		found bool
+		guid  uint64
+		r, s  int
+	}
+
+findCandidate:
+	for r := len(receiver) - 1; r >= 0; r-- {
+		for s := len(sender) - 1; s >= 0; s-- {
+			if sender[s].GetGuid() == receiver[r].GetGuid() {
+				mrcaCandidate.guid = sender[s].GetGuid()
+				mrcaCandidate.s = s
+				mrcaCandidate.r = r
+				mrcaCandidate.found = true
+				break findCandidate
+			}
+		}
+	}
+
+	// handle failure cases
+	if !mrcaCandidate.found {
+		if len(sender) == 0 {
+			return nil, &ConflictNoSenderSnapshots{}
+		} else {
+			return nil, &ConflictNoCommonAncestor{
+				SortedSenderVersions:   sender,
+				SortedReceiverVersions: receiver,
+			}
+		}
+	} else if mrcaCandidate.r != len(receiver)-1 {
+		return nil, &ConflictDiverged{
+			SortedSenderVersions:   sender,
+			SortedReceiverVersions: receiver,
+			CommonAncestor:         sender[mrcaCandidate.s],
+			SenderOnly:             sender[mrcaCandidate.s+1:],
+			ReceiverOnly:           receiver[mrcaCandidate.r+1:],
+		}
+	}
+
+	// incPath is possible
+
+	// incPath must not contain bookmarks except initial one,
+	incPath = make([]*FilesystemVersion, 0, len(sender))
+	incPath = append(incPath, sender[mrcaCandidate.s])
+	// it's ok if incPath[0] is a bookmark, but not the subsequent ones in the incPath
+	for i := mrcaCandidate.s + 1; i < len(sender); i++ {
+		if sender[i].Type == FilesystemVersion_Snapshot && incPath[len(incPath)-1].Guid != sender[i].Guid {
+			incPath = append(incPath, sender[i])
+		}
+	}
+	if len(incPath) == 1 {
+		// nothing to do
+		return nil, &ConflictMostRecentSnapshotAlreadyPresent{
+			SortedSenderVersions:   sender,
+			SortedReceiverVersions: receiver,
+			CommonAncestor:         sender[mrcaCandidate.s],
+		}
+	}
+	return incPath, nil
+}

internal/replication/logic/diff/diff_test.go

@@ -0,0 +1,169 @@
+package diff
+
+import (
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	. "github.com/zrepl/zrepl/internal/replication/logic/pdu"
+)
+
+func fsvlist(fsv ...string) (r []*FilesystemVersion) {
+
+	r = make([]*FilesystemVersion, len(fsv))
+	for i, f := range fsv {
+
+		// parse the id from fsvlist. it is used to derive Guid,CreateTXG and Creation attrs
+		split := strings.Split(f, ",")
+		if len(split) != 2 {
+			panic("invalid fsv spec")
+		}
+		id, err := strconv.Atoi(split[1])
+		if err != nil {
+			panic(err)
+		}
+		creation := func(id int) string {
+			return FilesystemVersionCreation(time.Unix(0, 0).Add(time.Duration(id) * time.Second))
+		}
+		if strings.HasPrefix(f, "#") {
+			r[i] = &FilesystemVersion{
+				Name:      strings.TrimPrefix(f, "#"),
+				Type:      FilesystemVersion_Bookmark,
+				Guid:      uint64(id),
+				CreateTXG: uint64(id),
+				Creation:  creation(id),
+			}
+		} else if strings.HasPrefix(f, "@") {
+			r[i] = &FilesystemVersion{
+				Name:      strings.TrimPrefix(f, "@"),
+				Type:      FilesystemVersion_Snapshot,
+				Guid:      uint64(id),
+				CreateTXG: uint64(id),
+				Creation:  creation(id),
+			}
+		} else {
+			panic("invalid character")
+		}
+	}
+	return
+}
+
+func doTest(receiver, sender []*FilesystemVersion, validate func(incpath []*FilesystemVersion, conflict error)) {
+	p, err := IncrementalPath(receiver, sender)
+	validate(p, err)
+}
+
+func TestIncrementalPath_SnapshotsOnly(t *testing.T) {
+
+	l := fsvlist
+
+	// basic functionality
+	doTest(l("@a,1", "@b,2"), l("@a,1", "@b,2", "@c,3", "@d,4"), func(path []*FilesystemVersion, conflict error) {
+		assert.Equal(t, l("@b,2", "@c,3", "@d,4"), path)
+	})
+
+	// no common ancestor
+	doTest(l(), l("@a,1"), func(path []*FilesystemVersion, conflict error) {
+		assert.Nil(t, path)
+		ca, ok := conflict.(*ConflictNoCommonAncestor)
+		require.True(t, ok)
+		assert.Equal(t, l("@a,1"), ca.SortedSenderVersions)
+	})
+	doTest(l("@a,1", "@b,2"), l("@c,3", "@d,4"), func(path []*FilesystemVersion, conflict error) {
+		assert.Nil(t, path)
+		ca, ok := conflict.(*ConflictNoCommonAncestor)
+		require.True(t, ok)
+		assert.Equal(t, l("@a,1", "@b,2"), ca.SortedReceiverVersions)
+		assert.Equal(t, l("@c,3", "@d,4"), ca.SortedSenderVersions)
+	})
+
+	// divergence is detected
+	doTest(l("@a,1", "@b1,2"), l("@a,1", "@b2,3"), func(path []*FilesystemVersion, conflict error) {
+		assert.Nil(t, path)
+		cd, ok := conflict.(*ConflictDiverged)
+		require.True(t, ok)
+		assert.Equal(t, l("@a,1")[0], cd.CommonAncestor)
+		assert.Equal(t, l("@b1,2"), cd.ReceiverOnly)
+		assert.Equal(t, l("@b2,3"), cd.SenderOnly)
+	})
+
+	// gaps before most recent common ancestor do not matter
+	doTest(l("@a,1", "@b,2", "@c,3"), l("@a,1", "@c,3", "@d,4"), func(path []*FilesystemVersion, conflict error) {
+		assert.Equal(t, l("@c,3", "@d,4"), path)
+	})
+
+	// nothing to do if fully shared history
+	doTest(l("@a,1", "@b,2"), l("@a,1", "@b,2"), func(incpath []*FilesystemVersion, conflict error) {
+		assert.Nil(t, incpath)
+		assert.NotNil(t, conflict)
+		_, ok := conflict.(*ConflictMostRecentSnapshotAlreadyPresent)
+		assert.True(t, ok)
+	})
+
+	// ...but it's sufficient if the most recent snapshot is present
+	doTest(l("@c,3"), l("@a,1", "@b,2", "@c,3"), func(path []*FilesystemVersion, conflict error) {
+		assert.Nil(t, path)
+		_, ok := conflict.(*ConflictMostRecentSnapshotAlreadyPresent)
+		assert.True(t, ok)
+	})
+
+	// no sender snapshots errors: empty receiver
+	doTest(l(), l(), func(incpath []*FilesystemVersion, conflict error) {
+		assert.Nil(t, incpath)
+		assert.NotNil(t, conflict)
+		t.Logf("%T", conflict)
+		_, ok := conflict.(*ConflictNoSenderSnapshots)
+		assert.True(t, ok)
+	})
+
+	// no sender snapshots errors: snapshots on receiver
+	doTest(l("@a,1"), l(), func(incpath []*FilesystemVersion, conflict error) {
+		assert.Nil(t, incpath)
+		assert.NotNil(t, conflict)
+		t.Logf("%T", conflict)
+		_, ok := conflict.(*ConflictNoSenderSnapshots)
+		assert.True(t, ok)
+	})
+
+}
+
+func TestIncrementalPath_BookmarkSupport(t *testing.T) {
+	l := fsvlist
+
+	// bookmarks are used
+	doTest(l("@a,1"), l("#a,1", "@b,2"), func(path []*FilesystemVersion, conflict error) {
+		assert.Equal(t, l("#a,1", "@b,2"), path)
+	})
+
+	// bookmarks are stripped from IncrementalPath (cannot send incrementally)
+	doTest(l("@a,1"), l("#a,1", "#b,2", "@c,3"), func(path []*FilesystemVersion, conflict error) {
+		assert.Equal(t, l("#a,1", "@c,3"), path)
+	})
+
+	// test that snapshots are preferred over bookmarks in IncrementalPath
+	doTest(l("@a,1"), l("#a,1", "@a,1", "@b,2"), func(path []*FilesystemVersion, conflict error) {
+		assert.Equal(t, l("@a,1", "@b,2"), path)
+	})
+	doTest(l("@a,1"), l("@a,1", "#a,1", "@b,2"), func(path []*FilesystemVersion, conflict error) {
+		assert.Equal(t, l("@a,1", "@b,2"), path)
+	})
+
+	// test that receive-side bookmarks younger than the most recent common ancestor do not cause a conflict
+	doTest(l("@a,1", "#b,2"), l("@a,1", "@c,3"), func(path []*FilesystemVersion, conflict error) {
+		assert.NoError(t, conflict)
+		require.Len(t, path, 2)
+		assert.Equal(t, l("@a,1")[0], path[0])
+		assert.Equal(t, l("@c,3")[0], path[1])
+	})
+	doTest(l("#a,1"), l("@a,1", "@b,2"), func(path []*FilesystemVersion, conflict error) {
+		assert.Nil(t, path)
+		ca, ok := conflict.(*ConflictNoCommonAncestor)
+		require.True(t, ok)
+		assert.Equal(t, l(), ca.SortedReceiverVersions, "See comment in IncrementalPath() on why we don't include the boomkmark here")
+	})
+
+}

internal/replication/logic/initialreplicationautoresolution_enumer.go

@@ -0,0 +1,61 @@
+// Code generated by "enumer -type=InitialReplicationAutoResolution -trimprefix=InitialReplicationAutoResolution"; DO NOT EDIT.
+
+package logic
+
+import (
+	"fmt"
+)
+
+const (
+	_InitialReplicationAutoResolutionName_0 = "MostRecentAll"
+	_InitialReplicationAutoResolutionName_1 = "Fail"
+)
+
+var (
+	_InitialReplicationAutoResolutionIndex_0 = [...]uint8{0, 10, 13}
+	_InitialReplicationAutoResolutionIndex_1 = [...]uint8{0, 4}
+)
+
+func (i InitialReplicationAutoResolution) String() string {
+	switch {
+	case 1 <= i && i <= 2:
+		i -= 1
+		return _InitialReplicationAutoResolutionName_0[_InitialReplicationAutoResolutionIndex_0[i]:_InitialReplicationAutoResolutionIndex_0[i+1]]
+	case i == 4:
+		return _InitialReplicationAutoResolutionName_1
+	default:
+		return fmt.Sprintf("InitialReplicationAutoResolution(%d)", i)
+	}
+}
+
+var _InitialReplicationAutoResolutionValues = []InitialReplicationAutoResolution{1, 2, 4}
+
+var _InitialReplicationAutoResolutionNameToValueMap = map[string]InitialReplicationAutoResolution{
+	_InitialReplicationAutoResolutionName_0[0:10]:  1,
+	_InitialReplicationAutoResolutionName_0[10:13]: 2,
+	_InitialReplicationAutoResolutionName_1[0:4]:   4,
+}
+
+// InitialReplicationAutoResolutionString retrieves an enum value from the enum constants string name.
+// Throws an error if the param is not part of the enum.
+func InitialReplicationAutoResolutionString(s string) (InitialReplicationAutoResolution, error) {
+	if val, ok := _InitialReplicationAutoResolutionNameToValueMap[s]; ok {
+		return val, nil
+	}
+	return 0, fmt.Errorf("%s does not belong to InitialReplicationAutoResolution values", s)
+}
+
+// InitialReplicationAutoResolutionValues returns all values of the enum
+func InitialReplicationAutoResolutionValues() []InitialReplicationAutoResolution {
+	return _InitialReplicationAutoResolutionValues
+}
+
+// IsAInitialReplicationAutoResolution returns "true" if the value is listed in the enum definition. "false" otherwise
+func (i InitialReplicationAutoResolution) IsAInitialReplicationAutoResolution() bool {
+	for _, v := range _InitialReplicationAutoResolutionValues {
+		if i == v {
+			return true
+		}
+	}
+	return false
+}

internal/replication/logic/pdu/pdu.proto

@@ -0,0 +1,137 @@
+syntax = "proto3";
+option go_package = ".;pdu";
+
+service Replication {
+  rpc Ping(PingReq) returns (PingRes);
+  rpc ListFilesystems(ListFilesystemReq) returns (ListFilesystemRes);
+  rpc ListFilesystemVersions(ListFilesystemVersionsReq)
+      returns (ListFilesystemVersionsRes);
+  rpc DestroySnapshots(DestroySnapshotsReq) returns (DestroySnapshotsRes);
+  rpc ReplicationCursor(ReplicationCursorReq) returns (ReplicationCursorRes);
+  rpc SendDry(SendReq) returns (SendRes);
+  rpc SendCompleted(SendCompletedReq) returns (SendCompletedRes);
+  // for Send and Recv, see package rpc
+}
+
+message ListFilesystemReq {}
+
+message ListFilesystemRes { repeated Filesystem Filesystems = 1; }
+
+message Filesystem {
+  string Path = 1;
+  string ResumeToken = 2;
+  bool IsPlaceholder = 3;
+}
+
+message ListFilesystemVersionsReq { string Filesystem = 1; }
+
+message ListFilesystemVersionsRes { repeated FilesystemVersion Versions = 1; }
+
+message FilesystemVersion {
+  enum VersionType {
+    Snapshot = 0;
+    Bookmark = 1;
+  }
+  VersionType Type = 1;
+  string Name = 2;
+  uint64 Guid = 3;
+  uint64 CreateTXG = 4;
+  string Creation = 5; // RFC 3339
+}
+
+message SendReq {
+  string Filesystem = 1;
+  // May be empty / null to request a full transfer of To
+  FilesystemVersion From = 2;
+  FilesystemVersion To = 3;
+
+  // If ResumeToken is not empty, the resume token that CAN be used for 'zfs
+  // send' by the sender. The sender MUST indicate use of ResumeToken in the
+  // reply message SendRes.UsedResumeToken If it does not work, the sender
+  // SHOULD clear the resume token on their side and use From and To instead If
+  // ResumeToken is not empty, the GUIDs of From and To MUST correspond to those
+  // encoded in the ResumeToken. Otherwise, the Sender MUST return an error.
+  string ResumeToken = 4;
+
+  ReplicationConfig ReplicationConfig = 6;
+}
+
+message ReplicationConfig {
+  ReplicationConfigProtection protection = 1;
+}
+
+
+message ReplicationConfigProtection {
+  ReplicationGuaranteeKind Initial = 1;
+  ReplicationGuaranteeKind Incremental = 2;
+}
+
+enum ReplicationGuaranteeKind {
+  GuaranteeInvalid = 0;
+  GuaranteeResumability = 1;
+  GuaranteeIncrementalReplication = 2;
+  GuaranteeNothing = 3; 
+}
+
+message Property {
+  string Name = 1;
+  string Value = 2;
+}
+
+message SendRes {
+  // Whether the resume token provided in the request has been used or not.
+  // If the SendReq.ResumeToken == "", this field MUST be false.
+  bool UsedResumeToken = 1;
+
+  // Expected stream size determined by dry run, not exact.
+  // 0 indicates that for the given SendReq, no size estimate could be made.
+  uint64 ExpectedSize = 2;
+}
+
+message SendCompletedReq {
+  SendReq OriginalReq = 2;
+}
+
+message SendCompletedRes {}
+
+message ReceiveReq {
+  string Filesystem = 1;
+  FilesystemVersion To = 2;
+
+  // If true, the receiver should clear the resume token before performing the
+  // zfs recv of the stream in the request
+  bool ClearResumeToken = 3;
+
+  ReplicationConfig ReplicationConfig = 4;
+}
+
+message ReceiveRes {}
+
+message DestroySnapshotsReq {
+  string Filesystem = 1;
+  // Path to filesystem, snapshot or bookmark to be destroyed
+  repeated FilesystemVersion Snapshots = 2;
+}
+
+message DestroySnapshotRes {
+  FilesystemVersion Snapshot = 1;
+  string Error = 2;
+}
+
+message DestroySnapshotsRes { repeated DestroySnapshotRes Results = 1; }
+
+message ReplicationCursorReq { string Filesystem = 1; }
+
+message ReplicationCursorRes {
+  oneof Result {
+    uint64 Guid = 1;
+    bool Notexist = 2;
+  }
+}
+
+message PingReq { string Message = 1; }
+
+message PingRes {
+  // Echo must be PingReq.Message
+  string Echo = 1;
+}

internal/replication/logic/pdu/pdu_extras.go

@@ -0,0 +1,92 @@
+package pdu
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/zrepl/zrepl/internal/zfs"
+)
+
+func (v *FilesystemVersion) GetRelName() string {
+	zv, err := v.ZFSFilesystemVersion()
+	if err != nil {
+		return ""
+	}
+	return zv.String()
+}
+
+func (v *FilesystemVersion) RelName() string {
+	zv, err := v.ZFSFilesystemVersion()
+	if err != nil {
+		panic(err)
+	}
+	return zv.String()
+}
+
+func (v FilesystemVersion_VersionType) ZFSVersionType() zfs.VersionType {
+	switch v {
+	case FilesystemVersion_Snapshot:
+		return zfs.Snapshot
+	case FilesystemVersion_Bookmark:
+		return zfs.Bookmark
+	default:
+		panic(fmt.Sprintf("unexpected v.Type %#v", v))
+	}
+}
+
+func FilesystemVersionFromZFS(fsv *zfs.FilesystemVersion) *FilesystemVersion {
+	var t FilesystemVersion_VersionType
+	switch fsv.Type {
+	case zfs.Bookmark:
+		t = FilesystemVersion_Bookmark
+	case zfs.Snapshot:
+		t = FilesystemVersion_Snapshot
+	default:
+		panic("unknown fsv.Type: " + fsv.Type)
+	}
+	return &FilesystemVersion{
+		Type:      t,
+		Name:      fsv.Name,
+		Guid:      fsv.Guid,
+		CreateTXG: fsv.CreateTXG,
+		Creation:  fsv.Creation.Format(time.RFC3339),
+	}
+}
+
+func FilesystemVersionCreation(t time.Time) string {
+	return t.Format(time.RFC3339)
+}
+
+func (v *FilesystemVersion) CreationAsTime() (time.Time, error) {
+	return time.Parse(time.RFC3339, v.Creation)
+}
+
+// implement fsfsm.FilesystemVersion
+func (v *FilesystemVersion) SnapshotTime() time.Time {
+	t, err := v.CreationAsTime()
+	if err != nil {
+		panic(err) // FIXME
+	}
+	return t
+}
+
+func (v *FilesystemVersion) ZFSFilesystemVersion() (*zfs.FilesystemVersion, error) {
+	ct, err := v.CreationAsTime()
+	if err != nil {
+		return nil, err
+	}
+	return &zfs.FilesystemVersion{
+		Type:      v.Type.ZFSVersionType(),
+		Name:      v.Name,
+		Guid:      v.Guid,
+		CreateTXG: v.CreateTXG,
+		Creation:  ct,
+	}, nil
+}
+
+func ReplicationConfigProtectionWithKind(both ReplicationGuaranteeKind) *ReplicationConfigProtection {
+	return &ReplicationConfigProtection{
+		Initial:     both,
+		Incremental: both,
+	}
+}

internal/replication/logic/pdu/pdu_grpc.pb.go

@@ -0,0 +1,349 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.5.1
+// - protoc             v5.28.0
+// source: pdu.proto
+
+package pdu
+
+import (
+	context "context"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.64.0 or later.
+const _ = grpc.SupportPackageIsVersion9
+
+const (
+	Replication_Ping_FullMethodName                   = "/Replication/Ping"
+	Replication_ListFilesystems_FullMethodName        = "/Replication/ListFilesystems"
+	Replication_ListFilesystemVersions_FullMethodName = "/Replication/ListFilesystemVersions"
+	Replication_DestroySnapshots_FullMethodName       = "/Replication/DestroySnapshots"
+	Replication_ReplicationCursor_FullMethodName      = "/Replication/ReplicationCursor"
+	Replication_SendDry_FullMethodName                = "/Replication/SendDry"
+	Replication_SendCompleted_FullMethodName          = "/Replication/SendCompleted"
+)
+
+// ReplicationClient is the client API for Replication service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+type ReplicationClient interface {
+	Ping(ctx context.Context, in *PingReq, opts ...grpc.CallOption) (*PingRes, error)
+	ListFilesystems(ctx context.Context, in *ListFilesystemReq, opts ...grpc.CallOption) (*ListFilesystemRes, error)
+	ListFilesystemVersions(ctx context.Context, in *ListFilesystemVersionsReq, opts ...grpc.CallOption) (*ListFilesystemVersionsRes, error)
+	DestroySnapshots(ctx context.Context, in *DestroySnapshotsReq, opts ...grpc.CallOption) (*DestroySnapshotsRes, error)
+	ReplicationCursor(ctx context.Context, in *ReplicationCursorReq, opts ...grpc.CallOption) (*ReplicationCursorRes, error)
+	SendDry(ctx context.Context, in *SendReq, opts ...grpc.CallOption) (*SendRes, error)
+	SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error)
+}
+
+type replicationClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewReplicationClient(cc grpc.ClientConnInterface) ReplicationClient {
+	return &replicationClient{cc}
+}
+
+func (c *replicationClient) Ping(ctx context.Context, in *PingReq, opts ...grpc.CallOption) (*PingRes, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(PingRes)
+	err := c.cc.Invoke(ctx, Replication_Ping_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) ListFilesystems(ctx context.Context, in *ListFilesystemReq, opts ...grpc.CallOption) (*ListFilesystemRes, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ListFilesystemRes)
+	err := c.cc.Invoke(ctx, Replication_ListFilesystems_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) ListFilesystemVersions(ctx context.Context, in *ListFilesystemVersionsReq, opts ...grpc.CallOption) (*ListFilesystemVersionsRes, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ListFilesystemVersionsRes)
+	err := c.cc.Invoke(ctx, Replication_ListFilesystemVersions_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) DestroySnapshots(ctx context.Context, in *DestroySnapshotsReq, opts ...grpc.CallOption) (*DestroySnapshotsRes, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(DestroySnapshotsRes)
+	err := c.cc.Invoke(ctx, Replication_DestroySnapshots_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) ReplicationCursor(ctx context.Context, in *ReplicationCursorReq, opts ...grpc.CallOption) (*ReplicationCursorRes, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(ReplicationCursorRes)
+	err := c.cc.Invoke(ctx, Replication_ReplicationCursor_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) SendDry(ctx context.Context, in *SendReq, opts ...grpc.CallOption) (*SendRes, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(SendRes)
+	err := c.cc.Invoke(ctx, Replication_SendDry_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+func (c *replicationClient) SendCompleted(ctx context.Context, in *SendCompletedReq, opts ...grpc.CallOption) (*SendCompletedRes, error) {
+	cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
+	out := new(SendCompletedRes)
+	err := c.cc.Invoke(ctx, Replication_SendCompleted_FullMethodName, in, out, cOpts...)
+	if err != nil {
+		return nil, err
+	}
+	return out, nil
+}
+
+// ReplicationServer is the server API for Replication service.
+// All implementations must embed UnimplementedReplicationServer
+// for forward compatibility.
+type ReplicationServer interface {
+	Ping(context.Context, *PingReq) (*PingRes, error)
+	ListFilesystems(context.Context, *ListFilesystemReq) (*ListFilesystemRes, error)
+	ListFilesystemVersions(context.Context, *ListFilesystemVersionsReq) (*ListFilesystemVersionsRes, error)
+	DestroySnapshots(context.Context, *DestroySnapshotsReq) (*DestroySnapshotsRes, error)
+	ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error)
+	SendDry(context.Context, *SendReq) (*SendRes, error)
+	SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error)
+	mustEmbedUnimplementedReplicationServer()
+}
+
+// UnimplementedReplicationServer must be embedded to have
+// forward compatible implementations.
+//
+// NOTE: this should be embedded by value instead of pointer to avoid a nil
+// pointer dereference when methods are called.
+type UnimplementedReplicationServer struct{}
+
+func (UnimplementedReplicationServer) Ping(context.Context, *PingReq) (*PingRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented")
+}
+func (UnimplementedReplicationServer) ListFilesystems(context.Context, *ListFilesystemReq) (*ListFilesystemRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ListFilesystems not implemented")
+}
+func (UnimplementedReplicationServer) ListFilesystemVersions(context.Context, *ListFilesystemVersionsReq) (*ListFilesystemVersionsRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ListFilesystemVersions not implemented")
+}
+func (UnimplementedReplicationServer) DestroySnapshots(context.Context, *DestroySnapshotsReq) (*DestroySnapshotsRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method DestroySnapshots not implemented")
+}
+func (UnimplementedReplicationServer) ReplicationCursor(context.Context, *ReplicationCursorReq) (*ReplicationCursorRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method ReplicationCursor not implemented")
+}
+func (UnimplementedReplicationServer) SendDry(context.Context, *SendReq) (*SendRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method SendDry not implemented")
+}
+func (UnimplementedReplicationServer) SendCompleted(context.Context, *SendCompletedReq) (*SendCompletedRes, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method SendCompleted not implemented")
+}
+func (UnimplementedReplicationServer) mustEmbedUnimplementedReplicationServer() {}
+func (UnimplementedReplicationServer) testEmbeddedByValue()                     {}
+
+// UnsafeReplicationServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to ReplicationServer will
+// result in compilation errors.
+type UnsafeReplicationServer interface {
+	mustEmbedUnimplementedReplicationServer()
+}
+
+func RegisterReplicationServer(s grpc.ServiceRegistrar, srv ReplicationServer) {
+	// If the following call pancis, it indicates UnimplementedReplicationServer was
+	// embedded by pointer and is nil.  This will cause panics if an
+	// unimplemented method is ever invoked, so we test this at initialization
+	// time to prevent it from happening at runtime later due to I/O.
+	if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
+		t.testEmbeddedByValue()
+	}
+	s.RegisterService(&Replication_ServiceDesc, srv)
+}
+
+func _Replication_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(PingReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).Ping(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: Replication_Ping_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).Ping(ctx, req.(*PingReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_ListFilesystems_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ListFilesystemReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).ListFilesystems(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: Replication_ListFilesystems_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).ListFilesystems(ctx, req.(*ListFilesystemReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_ListFilesystemVersions_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ListFilesystemVersionsReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).ListFilesystemVersions(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: Replication_ListFilesystemVersions_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).ListFilesystemVersions(ctx, req.(*ListFilesystemVersionsReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_DestroySnapshots_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(DestroySnapshotsReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).DestroySnapshots(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: Replication_DestroySnapshots_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).DestroySnapshots(ctx, req.(*DestroySnapshotsReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_ReplicationCursor_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(ReplicationCursorReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).ReplicationCursor(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: Replication_ReplicationCursor_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).ReplicationCursor(ctx, req.(*ReplicationCursorReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_SendDry_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(SendReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).SendDry(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: Replication_SendDry_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).SendDry(ctx, req.(*SendReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+func _Replication_SendCompleted_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(SendCompletedReq)
+	if err := dec(in); err != nil {
+		return nil, err
+	}
+	if interceptor == nil {
+		return srv.(ReplicationServer).SendCompleted(ctx, in)
+	}
+	info := &grpc.UnaryServerInfo{
+		Server:     srv,
+		FullMethod: Replication_SendCompleted_FullMethodName,
+	}
+	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
+		return srv.(ReplicationServer).SendCompleted(ctx, req.(*SendCompletedReq))
+	}
+	return interceptor(ctx, in, info, handler)
+}
+
+// Replication_ServiceDesc is the grpc.ServiceDesc for Replication service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var Replication_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "Replication",
+	HandlerType: (*ReplicationServer)(nil),
+	Methods: []grpc.MethodDesc{
+		{
+			MethodName: "Ping",
+			Handler:    _Replication_Ping_Handler,
+		},
+		{
+			MethodName: "ListFilesystems",
+			Handler:    _Replication_ListFilesystems_Handler,
+		},
+		{
+			MethodName: "ListFilesystemVersions",
+			Handler:    _Replication_ListFilesystemVersions_Handler,
+		},
+		{
+			MethodName: "DestroySnapshots",
+			Handler:    _Replication_DestroySnapshots_Handler,
+		},
+		{
+			MethodName: "ReplicationCursor",
+			Handler:    _Replication_ReplicationCursor_Handler,
+		},
+		{
+			MethodName: "SendDry",
+			Handler:    _Replication_SendDry_Handler,
+		},
+		{
+			MethodName: "SendCompleted",
+			Handler:    _Replication_SendCompleted_Handler,
+		},
+	},
+	Streams:  []grpc.StreamDesc{},
+	Metadata: "pdu.proto",
+}

internal/replication/logic/pdu/pdu_test.go

@@ -0,0 +1,69 @@
+package pdu
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFilesystemVersion_RelName(t *testing.T) {
+
+	type TestCase struct {
+		In    *FilesystemVersion
+		Out   string
+		Panic bool
+	}
+
+	creat := FilesystemVersionCreation(time.Now())
+	tcs := []TestCase{
+		{
+			In: &FilesystemVersion{
+				Type:     FilesystemVersion_Snapshot,
+				Name:     "foobar",
+				Creation: creat,
+			},
+			Out: "@foobar",
+		},
+		{
+			In: &FilesystemVersion{
+				Type:     FilesystemVersion_Bookmark,
+				Name:     "foobar",
+				Creation: creat,
+			},
+			Out: "#foobar",
+		},
+		{
+			In: &FilesystemVersion{
+				Type:     2342,
+				Name:     "foobar",
+				Creation: creat,
+			},
+			Panic: true,
+		},
+	}
+
+	for _, tc := range tcs {
+		if tc.Panic {
+			assert.Panics(t, func() {
+				tc.In.RelName()
+			})
+		} else {
+			o := tc.In.RelName()
+			assert.Equal(t, tc.Out, o)
+		}
+	}
+
+}
+
+func TestFilesystemVersion_ZFSFilesystemVersion(t *testing.T) {
+
+	empty := &FilesystemVersion{}
+	_, err := empty.ZFSFilesystemVersion()
+	assert.Error(t, err)
+
+	dateInvalid := &FilesystemVersion{Creation: "foobar"}
+	_, err = dateInvalid.ZFSFilesystemVersion()
+	assert.Error(t, err)
+
+}

internal/replication/logic/replication_logic.go

@@ -0,0 +1,656 @@
+package logic
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sync"
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/zrepl/zrepl/internal/daemon/logging/trace"
+
+	"github.com/zrepl/zrepl/internal/logger"
+	"github.com/zrepl/zrepl/internal/replication/driver"
+	. "github.com/zrepl/zrepl/internal/replication/logic/diff"
+	"github.com/zrepl/zrepl/internal/replication/logic/pdu"
+	"github.com/zrepl/zrepl/internal/replication/report"
+	"github.com/zrepl/zrepl/internal/util/bytecounter"
+	"github.com/zrepl/zrepl/internal/util/chainlock"
+	"github.com/zrepl/zrepl/internal/util/semaphore"
+	"github.com/zrepl/zrepl/internal/zfs"
+)
+
+// Endpoint represents one side of the replication.
+//
+// An endpoint is either in Sender or Receiver mode, represented by the correspondingly
+// named interfaces defined in this package.
+type Endpoint interface {
+	// Does not include placeholder filesystems
+	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
+}
+
+type Sender interface {
+	Endpoint
+	// If a non-nil io.ReadCloser is returned, it is guaranteed to be closed before
+	// any next call to the parent github.com/zrepl/zrepl/replication.Endpoint.
+	// If the send request is for dry run the io.ReadCloser will be nil
+	Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error)
+	SendDry(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, error)
+	SendCompleted(ctx context.Context, r *pdu.SendCompletedReq) (*pdu.SendCompletedRes, error)
+	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
+}
+
+type Receiver interface {
+	Endpoint
+	// Receive sends r and sendStream (the latter containing a ZFS send stream)
+	// to the parent github.com/zrepl/zrepl/replication.Endpoint.
+	Receive(ctx context.Context, req *pdu.ReceiveReq, receive io.ReadCloser) (*pdu.ReceiveRes, error)
+}
+
+type Planner struct {
+	sender   Sender
+	receiver Receiver
+	policy   PlannerPolicy
+
+	promSecsPerState    *prometheus.HistogramVec // labels: state
+	promBytesReplicated *prometheus.CounterVec   // labels: filesystem
+}
+
+func (p *Planner) Plan(ctx context.Context) ([]driver.FS, error) {
+	fss, err := p.doPlanning(ctx)
+	if err != nil {
+		return nil, err
+	}
+	dfss := make([]driver.FS, len(fss))
+	for i := range dfss {
+		dfss[i] = fss[i]
+	}
+	return dfss, nil
+}
+
+func (p *Planner) WaitForConnectivity(ctx context.Context) error {
+	var wg sync.WaitGroup
+	doPing := func(endpoint Endpoint, errOut *error) {
+		defer wg.Done()
+		ctx, endTask := trace.WithTaskFromStack(ctx)
+		defer endTask()
+		err := endpoint.WaitForConnectivity(ctx)
+		if err != nil {
+			*errOut = err
+		} else {
+			*errOut = nil
+		}
+	}
+	wg.Add(2)
+	var senderErr, receiverErr error
+	go doPing(p.sender, &senderErr)
+	go doPing(p.receiver, &receiverErr)
+	wg.Wait()
+	if senderErr == nil && receiverErr == nil {
+		return nil
+	} else if senderErr != nil && receiverErr != nil {
+		if senderErr.Error() == receiverErr.Error() {
+			return fmt.Errorf("sender and receiver are not reachable: %s", senderErr.Error())
+		} else {
+			return fmt.Errorf("sender and receiver are not reachable:\n  sender: %s\n  receiver: %s", senderErr, receiverErr)
+		}
+	} else {
+		var side string
+		var err *error
+		if senderErr != nil {
+			side = "sender"
+			err = &senderErr
+		} else {
+			side = "receiver"
+			err = &receiverErr
+		}
+		return fmt.Errorf("%s is not reachable: %s", side, *err)
+	}
+}
+
+type Filesystem struct {
+	sender   Sender
+	receiver Receiver
+	policy   PlannerPolicy // immutable, it's .ReplicationConfig member is a pointer and copied into messages
+
+	Path                 string             // compat
+	receiverFS, senderFS *pdu.Filesystem    // receiverFS may be nil, senderFS never nil
+	promBytesReplicated  prometheus.Counter // compat
+
+	sizeEstimateRequestSem *semaphore.S
+}
+
+func (f *Filesystem) EqualToPreviousAttempt(other driver.FS) bool {
+	g, ok := other.(*Filesystem)
+	if !ok {
+		return false
+	}
+	// TODO: use GUIDs (issued by zrepl, not those from ZFS)
+	return f.Path == g.Path
+}
+
+func (f *Filesystem) PlanFS(ctx context.Context) ([]driver.Step, error) {
+	steps, err := f.doPlanning(ctx)
+	if err != nil {
+		return nil, err
+	}
+	dsteps := make([]driver.Step, len(steps))
+	for i := range dsteps {
+		dsteps[i] = steps[i]
+	}
+	return dsteps, nil
+}
+func (f *Filesystem) ReportInfo() *report.FilesystemInfo {
+	return &report.FilesystemInfo{Name: f.Path} // FIXME compat name
+}
+
+type Step struct {
+	sender   Sender
+	receiver Receiver
+
+	parent      *Filesystem
+	from, to    *pdu.FilesystemVersion // from may be nil, indicating full send
+	resumeToken string                 // empty means no resume token shall be used
+
+	expectedSize uint64 // 0 means no size estimate present / possible
+
+	// byteCounter is nil initially, and set later in Step.doReplication
+	// => concurrent read of that pointer from Step.ReportInfo must be protected
+	byteCounter    bytecounter.ReadCloser
+	byteCounterMtx chainlock.L
+}
+
+func (s *Step) TargetEquals(other driver.Step) bool {
+	t, ok := other.(*Step)
+	if !ok {
+		return false
+	}
+	if !s.parent.EqualToPreviousAttempt(t.parent) {
+		panic("Step interface promise broken: parent filesystems must be same")
+	}
+	return s.from.GetGuid() == t.from.GetGuid() &&
+		s.to.GetGuid() == t.to.GetGuid()
+}
+
+func (s *Step) TargetDate() time.Time {
+	return s.to.SnapshotTime() // FIXME compat name
+}
+
+func (s *Step) Step(ctx context.Context) error {
+	return s.doReplication(ctx)
+}
+
+func (s *Step) ReportInfo() *report.StepInfo {
+
+	// get current byteCounter value
+	var byteCounter uint64
+	s.byteCounterMtx.Lock()
+	if s.byteCounter != nil {
+		byteCounter = s.byteCounter.Count()
+	}
+	s.byteCounterMtx.Unlock()
+
+	from := ""
+	if s.from != nil {
+		from = s.from.RelName()
+	}
+	return &report.StepInfo{
+		From:            from,
+		To:              s.to.RelName(),
+		Resumed:         s.resumeToken != "",
+		BytesExpected:   s.expectedSize,
+		BytesReplicated: byteCounter,
+	}
+}
+
+// caller must ensure policy.Validate() == nil
+func NewPlanner(secsPerState *prometheus.HistogramVec, bytesReplicated *prometheus.CounterVec, sender Sender, receiver Receiver, policy PlannerPolicy) *Planner {
+	if err := policy.Validate(); err != nil {
+		panic(err)
+	}
+	return &Planner{
+		sender:              sender,
+		receiver:            receiver,
+		policy:              policy,
+		promSecsPerState:    secsPerState,
+		promBytesReplicated: bytesReplicated,
+	}
+}
+
+func tryAutoresolveConflict(conflict error, policy ConflictResolution) (path []*pdu.FilesystemVersion, reason error) {
+
+	if _, ok := conflict.(*ConflictMostRecentSnapshotAlreadyPresent); ok {
+		// replicatoin is a no-op
+		return nil, nil
+	}
+
+	if noCommonAncestor, ok := conflict.(*ConflictNoCommonAncestor); ok {
+		if len(noCommonAncestor.SortedReceiverVersions) == 0 {
+
+			if len(noCommonAncestor.SortedSenderVersions) == 0 {
+				return nil, fmt.Errorf("no snapshots available on sender side")
+			}
+
+			switch policy.InitialReplication {
+
+			case InitialReplicationAutoResolutionMostRecent:
+
+				var mostRecentSnap *pdu.FilesystemVersion
+				for n := len(noCommonAncestor.SortedSenderVersions) - 1; n >= 0; n-- {
+					if noCommonAncestor.SortedSenderVersions[n].Type == pdu.FilesystemVersion_Snapshot {
+						mostRecentSnap = noCommonAncestor.SortedSenderVersions[n]
+						break
+					}
+				}
+				return []*pdu.FilesystemVersion{nil, mostRecentSnap}, nil
+
+			case InitialReplicationAutoResolutionAll:
+
+				path = append(path, nil)
+
+				for n := 0; n < len(noCommonAncestor.SortedSenderVersions); n++ {
+					if noCommonAncestor.SortedSenderVersions[n].Type == pdu.FilesystemVersion_Snapshot {
+						path = append(path, noCommonAncestor.SortedSenderVersions[n])
+					}
+				}
+				return path, nil
+
+			case InitialReplicationAutoResolutionFail:
+
+				return nil, fmt.Errorf("automatic conflict resolution for initial replication is disabled in config")
+
+			default:
+				panic(fmt.Sprintf("unimplemented: %#v", policy.InitialReplication))
+			}
+		}
+	}
+	return nil, conflict
+}
+
+func (p *Planner) doPlanning(ctx context.Context) ([]*Filesystem, error) {
+
+	log := getLogger(ctx)
+
+	log.Info("start planning")
+
+	slfssres, err := p.sender.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
+	if err != nil {
+		log.WithError(err).WithField("errType", fmt.Sprintf("%T", err)).Error("error listing sender filesystems")
+		return nil, err
+	}
+	sfss := slfssres.GetFilesystems()
+
+	rlfssres, err := p.receiver.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
+	if err != nil {
+		log.WithError(err).WithField("errType", fmt.Sprintf("%T", err)).Error("error listing receiver filesystems")
+		return nil, err
+	}
+	rfss := rlfssres.GetFilesystems()
+
+	sizeEstimateRequestSem := semaphore.New(int64(p.policy.SizeEstimationConcurrency))
+
+	q := make([]*Filesystem, 0, len(sfss))
+	for _, fs := range sfss {
+
+		var receiverFS *pdu.Filesystem
+		for _, rfs := range rfss {
+			if rfs.Path == fs.Path {
+				receiverFS = rfs
+			}
+		}
+
+		var ctr prometheus.Counter
+		if p.promBytesReplicated != nil {
+			ctr = p.promBytesReplicated.WithLabelValues(fs.Path)
+		}
+
+		q = append(q, &Filesystem{
+			sender:                 p.sender,
+			receiver:               p.receiver,
+			policy:                 p.policy,
+			Path:                   fs.Path,
+			senderFS:               fs,
+			receiverFS:             receiverFS,
+			promBytesReplicated:    ctr,
+			sizeEstimateRequestSem: sizeEstimateRequestSem,
+		})
+	}
+
+	return q, nil
+}
+
+func (fs *Filesystem) doPlanning(ctx context.Context) ([]*Step, error) {
+
+	log := func(ctx context.Context) logger.Logger {
+		return getLogger(ctx).WithField("filesystem", fs.Path)
+	}
+
+	log(ctx).Debug("assessing filesystem")
+
+	if fs.senderFS.IsPlaceholder {
+		log(ctx).Debug("sender filesystem is placeholder")
+		if fs.receiverFS != nil {
+			if fs.receiverFS.IsPlaceholder {
+				// all good, fall through
+				log(ctx).Debug("receiver filesystem is placeholder")
+			} else {
+				err := fmt.Errorf("sender filesystem is placeholder, but receiver filesystem is not")
+				log(ctx).Error(err.Error())
+				return nil, err
+			}
+		}
+		log(ctx).Debug("no steps required for replicating placeholders, the endpoint.Receiver will create a placeholder when we receive the first non-placeholder child filesystem")
+		return nil, nil
+	}
+
+	sfsvsres, err := fs.sender.ListFilesystemVersions(ctx, &pdu.ListFilesystemVersionsReq{Filesystem: fs.Path})
+	if err != nil {
+		log(ctx).WithError(err).Error("cannot get remote filesystem versions")
+		return nil, err
+	}
+	sfsvs := sfsvsres.GetVersions()
+
+	if len(sfsvs) < 1 {
+		err := errors.New("sender does not have any versions")
+		log(ctx).Error(err.Error())
+		return nil, err
+	}
+
+	var rfsvs []*pdu.FilesystemVersion
+	if fs.receiverFS != nil && !fs.receiverFS.GetIsPlaceholder() {
+		rfsvsres, err := fs.receiver.ListFilesystemVersions(ctx, &pdu.ListFilesystemVersionsReq{Filesystem: fs.Path})
+		if err != nil {
+			log(ctx).WithError(err).Error("receiver error")
+			return nil, err
+		}
+		rfsvs = rfsvsres.GetVersions()
+	} else {
+		rfsvs = []*pdu.FilesystemVersion{}
+	}
+
+	var resumeToken *zfs.ResumeToken
+	var resumeTokenRaw string
+	if fs.receiverFS != nil && fs.receiverFS.ResumeToken != "" {
+		resumeTokenRaw = fs.receiverFS.ResumeToken // shadow
+		log(ctx).WithField("receiverFS.ResumeToken", resumeTokenRaw).Debug("decode receiver fs resume token")
+		resumeToken, err = zfs.ParseResumeToken(ctx, resumeTokenRaw) // shadow
+		if err != nil {
+			// TODO in theory, we could do replication without resume token, but that would mean that
+			// we need to discard the resumable state on the receiver's side.
+			// Would be easy by setting UsedResumeToken=false in the RecvReq ...
+			// FIXME / CHECK semantics UsedResumeToken if SendReq.ResumeToken == ""
+			log(ctx).WithError(err).Error("cannot decode resume token, aborting")
+			return nil, err
+		}
+		log(ctx).WithField("token", resumeToken).Debug("decode resume token")
+	}
+
+	var steps []*Step
+	// build the list of replication steps
+	//
+	// prefer to resume any started replication instead of starting over with a normal IncrementalPath
+	//
+	// look for the step encoded in the resume token in the sender's version
+	// if we find that step:
+	//   1. use it as first step (including resume token)
+	//   2. compute subsequent steps by computing incremental path from the token.To version on
+	//      ...
+	//      that's actually equivalent to simply cutting off earlier versions from rfsvs and sfsvs
+	if resumeToken != nil {
+
+		sfsvs := SortVersionListByCreateTXGThenBookmarkLTSnapshot(sfsvs)
+
+		var fromVersion, toVersion *pdu.FilesystemVersion
+		var toVersionIdx int
+		for idx, sfsv := range sfsvs {
+			if resumeToken.HasFromGUID && sfsv.Guid == resumeToken.FromGUID {
+				if fromVersion != nil && fromVersion.Type == pdu.FilesystemVersion_Snapshot {
+					// prefer snapshots over bookmarks for size estimation
+				} else {
+					fromVersion = sfsv
+				}
+			}
+			if resumeToken.HasToGUID && sfsv.Guid == resumeToken.ToGUID && sfsv.Type == pdu.FilesystemVersion_Snapshot {
+				// `toversion` must always be a snapshot
+				toVersion, toVersionIdx = sfsv, idx
+			}
+		}
+
+		if toVersion == nil {
+			return nil, fmt.Errorf("resume token `toguid` = %v not found on sender (`toname` = %q)", resumeToken.ToGUID, resumeToken.ToName)
+		} else if fromVersion == toVersion {
+			return nil, fmt.Errorf("resume token `fromguid` and `toguid` match same version on sener")
+		}
+		// fromVersion may be nil, toVersion is no nil, encryption matches
+		// good to go this one step!
+		resumeStep := &Step{
+			parent:   fs,
+			sender:   fs.sender,
+			receiver: fs.receiver,
+
+			from: fromVersion,
+			to:   toVersion,
+
+			resumeToken: resumeTokenRaw,
+		}
+
+		// by definition, the resume token _must_ be the receiver's most recent version, if they have any
+		// don't bother checking, zfs recv will produce an error if above assumption is wrong
+		//
+		// thus, subsequent steps are just incrementals on the sender's remaining _snapshots_ (not bookmarks)
+
+		var remainingSFSVs []*pdu.FilesystemVersion
+		for _, sfsv := range sfsvs[toVersionIdx:] {
+			if sfsv.Type == pdu.FilesystemVersion_Snapshot {
+				remainingSFSVs = append(remainingSFSVs, sfsv)
+			}
+		}
+
+		steps = make([]*Step, 0, len(remainingSFSVs)) // shadow
+		steps = append(steps, resumeStep)
+		for i := 0; i < len(remainingSFSVs)-1; i++ {
+			steps = append(steps, &Step{
+				parent:   fs,
+				sender:   fs.sender,
+				receiver: fs.receiver,
+				from:     remainingSFSVs[i],
+				to:       remainingSFSVs[i+1],
+			})
+		}
+	} else { // resumeToken == nil
+		path, conflict := IncrementalPath(rfsvs, sfsvs)
+		if conflict != nil {
+			updPath, updConflict := tryAutoresolveConflict(conflict, *fs.policy.ConflictResolution)
+			if updConflict == nil {
+				log(ctx).WithField("conflict", conflict).Info("conflict automatically resolved")
+
+			} else {
+				log(ctx).WithField("conflict", conflict).Error("cannot resolve conflict")
+			}
+			path, conflict = updPath, updConflict
+		}
+		if conflict != nil {
+			return nil, conflict
+		}
+		if len(path) == 0 {
+			steps = nil
+		} else if len(path) == 1 {
+			panic(fmt.Sprintf("len(path) must be two for incremental repl, and initial repl must start with nil, got path[0]=%#v", path[0]))
+		} else {
+			steps = make([]*Step, 0, len(path)) // shadow
+			for i := 0; i < len(path)-1; i++ {
+				steps = append(steps, &Step{
+					parent:   fs,
+					sender:   fs.sender,
+					receiver: fs.receiver,
+
+					from: path[i], // nil in case of initial repl
+					to:   path[i+1],
+				})
+			}
+		}
+	}
+
+	if len(steps) == 0 {
+		log(ctx).Info("planning determined that no replication steps are required")
+	}
+
+	log(ctx).Debug("compute send size estimate")
+	errs := make(chan error, len(steps))
+	fanOutCtx, fanOutCancel := context.WithCancel(ctx)
+	_, fanOutAdd, fanOutWait := trace.WithTaskGroup(fanOutCtx, "compute-size-estimate")
+	defer fanOutCancel()
+	for _, step := range steps {
+		step := step // local copy that is moved into the closure
+		fanOutAdd(func(ctx context.Context) {
+			// 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(ctx)
+			if err != nil {
+				fanOutCancel()
+				return
+			}
+			defer guard.Release()
+
+			err = step.updateSizeEstimate(ctx)
+			if err != nil {
+				log(ctx).WithError(err).WithField("step", step).Error("error computing size estimate")
+				fanOutCancel()
+			}
+			errs <- err
+		})
+	}
+	fanOutWait()
+	close(errs)
+	var significantErr error = nil
+	for err := range errs {
+		if err != nil {
+			if significantErr == nil || significantErr == context.Canceled {
+				significantErr = err
+			}
+		}
+	}
+	if significantErr != nil {
+		return nil, significantErr
+	}
+
+	log(ctx).Debug("filesystem planning finished")
+	return steps, nil
+}
+
+func (s *Step) updateSizeEstimate(ctx context.Context) error {
+
+	log := getLogger(ctx)
+
+	sr := s.buildSendRequest()
+
+	log.Debug("initiate dry run send request")
+	sres, err := s.sender.SendDry(ctx, sr)
+	if err != nil {
+		log.WithError(err).Error("dry run send request failed")
+		return err
+	}
+	if sres == nil {
+		err := fmt.Errorf("dry run send request returned nil send result")
+		log.Error(err.Error())
+		return err
+	}
+	s.expectedSize = sres.GetExpectedSize()
+	return nil
+}
+
+func (s *Step) buildSendRequest() (sr *pdu.SendReq) {
+	fs := s.parent.Path
+	sr = &pdu.SendReq{
+		Filesystem:        fs,
+		From:              s.from, // may be nil
+		To:                s.to,
+		ResumeToken:       s.resumeToken,
+		ReplicationConfig: s.parent.policy.ReplicationConfig,
+	}
+	return sr
+}
+
+func (s *Step) doReplication(ctx context.Context) error {
+
+	fs := s.parent.Path
+
+	log := getLogger(ctx).WithField("filesystem", fs)
+	sr := s.buildSendRequest()
+
+	log.Debug("initiate send request")
+	sres, stream, err := s.sender.Send(ctx, sr)
+	if err != nil {
+		log.WithError(err).Error("send request failed")
+		return err
+	}
+	if sres == nil {
+		err := fmt.Errorf("send request returned nil send result")
+		log.Error(err.Error())
+		return err
+	}
+	if stream == nil {
+		err := errors.New("send request did not return a stream, broken endpoint implementation")
+		return err
+	}
+	defer stream.Close()
+
+	// Install a byte counter to track progress + for status report
+	byteCountingStream := bytecounter.NewReadCloser(stream)
+	s.byteCounterMtx.Lock()
+	s.byteCounter = byteCountingStream
+	s.byteCounterMtx.Unlock()
+	defer func() {
+		defer s.byteCounterMtx.Lock().Unlock()
+		if s.parent.promBytesReplicated != nil {
+			s.parent.promBytesReplicated.Add(float64(s.byteCounter.Count()))
+		}
+	}()
+
+	rr := &pdu.ReceiveReq{
+		Filesystem:        fs,
+		To:                sr.GetTo(),
+		ClearResumeToken:  !sres.UsedResumeToken,
+		ReplicationConfig: s.parent.policy.ReplicationConfig,
+	}
+	log.Debug("initiate receive request")
+	_, err = s.receiver.Receive(ctx, rr, byteCountingStream)
+	if err != nil {
+		log.
+			WithError(err).
+			WithField("errType", fmt.Sprintf("%T", err)).
+			WithField("rr", fmt.Sprintf("%v", rr)).
+			Error("receive request failed (might also be error on sender)")
+		// This failure could be due to
+		// 	- an unexpected exit of ZFS on the sending side
+		//  - an unexpected exit of ZFS on the receiving side
+		//  - a connectivity issue
+		return err
+	}
+	log.Debug("receive finished")
+
+	log.Debug("tell sender replication completed")
+	_, err = s.sender.SendCompleted(ctx, &pdu.SendCompletedReq{
+		OriginalReq: sr,
+	})
+	if err != nil {
+		log.WithError(err).Error("error telling sender that replication completed successfully")
+		return err
+	}
+
+	return err
+}
+
+func (s *Step) String() string {
+	if s.from == nil { // FIXME: ZFS semantics are that to is nil on non-incremental send
+		return fmt.Sprintf("%s%s (full)", s.parent.Path, s.to.RelName())
+	} else {
+		return fmt.Sprintf("%s(%s => %s)", s.parent.Path, s.from.RelName(), s.to.RelName())
+	}
+}

internal/replication/logic/replication_logic_context.go

@@ -0,0 +1,12 @@
+package logic
+
+import (
+	"context"
+
+	"github.com/zrepl/zrepl/internal/daemon/logging"
+	"github.com/zrepl/zrepl/internal/logger"
+)
+
+func getLogger(ctx context.Context) logger.Logger {
+	return logging.GetLogger(ctx, logging.SubsysReplication)
+}

internal/replication/logic/replication_logic_policy.go

@@ -0,0 +1,111 @@
+package logic
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/go-playground/validator"
+	"github.com/pkg/errors"
+
+	"github.com/zrepl/zrepl/internal/config"
+	"github.com/zrepl/zrepl/internal/replication/logic/pdu"
+)
+
+//go:generate enumer -type=InitialReplicationAutoResolution -trimprefix=InitialReplicationAutoResolution
+type InitialReplicationAutoResolution uint32
+
+const (
+	InitialReplicationAutoResolutionMostRecent InitialReplicationAutoResolution = 1 << iota
+	InitialReplicationAutoResolutionAll
+	InitialReplicationAutoResolutionFail
+)
+
+var initialReplicationAutoResolutionConfigMap = map[InitialReplicationAutoResolution]string{
+	InitialReplicationAutoResolutionMostRecent: "most_recent",
+	InitialReplicationAutoResolutionAll:        "all",
+	InitialReplicationAutoResolutionFail:       "fail",
+}
+
+func InitialReplicationAutoResolutionFromConfig(in string) (InitialReplicationAutoResolution, error) {
+	for v, s := range initialReplicationAutoResolutionConfigMap {
+		if s == in {
+			return v, nil
+		}
+	}
+	l := make([]string, 0, len(initialReplicationAutoResolutionConfigMap))
+	for _, v := range InitialReplicationAutoResolutionValues() {
+		l = append(l, initialReplicationAutoResolutionConfigMap[v])
+	}
+	return 0, fmt.Errorf("invalid value %q, must be one of %s", in, strings.Join(l, ", "))
+}
+
+type ConflictResolution struct {
+	InitialReplication InitialReplicationAutoResolution
+}
+
+func (c *ConflictResolution) Validate() error {
+	if !c.InitialReplication.IsAInitialReplicationAutoResolution() {
+		return errors.Errorf("must be one of %s", InitialReplicationAutoResolutionValues())
+	}
+	return nil
+}
+
+func ConflictResolutionFromConfig(in *config.ConflictResolution) (*ConflictResolution, error) {
+
+	initialReplication, err := InitialReplicationAutoResolutionFromConfig(in.InitialReplication)
+	if err != nil {
+		return nil, errors.Errorf("field `initial_replication` is invalid: %q is not one of %v", in.InitialReplication, InitialReplicationAutoResolutionValues())
+	}
+
+	return &ConflictResolution{
+		InitialReplication: initialReplication,
+	}, nil
+}
+
+type PlannerPolicy struct {
+	ConflictResolution        *ConflictResolution    `validate:"ne=nil"`
+	ReplicationConfig         *pdu.ReplicationConfig `validate:"ne=nil"`
+	SizeEstimationConcurrency int                    `validate:"gte=1"`
+}
+
+var validate = validator.New()
+
+func (p PlannerPolicy) Validate() error {
+	if err := validate.Struct(p); err != nil {
+		return err
+	}
+	if err := p.ConflictResolution.Validate(); err != nil {
+		return err
+	}
+	return nil
+}
+
+func ReplicationConfigFromConfig(in *config.Replication) (*pdu.ReplicationConfig, error) {
+	initial, err := pduReplicationGuaranteeKindFromConfig(in.Protection.Initial)
+	if err != nil {
+		return nil, errors.Wrap(err, "field 'initial'")
+	}
+	incremental, err := pduReplicationGuaranteeKindFromConfig(in.Protection.Incremental)
+	if err != nil {
+		return nil, errors.Wrap(err, "field 'incremental'")
+	}
+	return &pdu.ReplicationConfig{
+		Protection: &pdu.ReplicationConfigProtection{
+			Initial:     initial,
+			Incremental: incremental,
+		},
+	}, nil
+}
+
+func pduReplicationGuaranteeKindFromConfig(in string) (k pdu.ReplicationGuaranteeKind, _ error) {
+	switch in {
+	case "guarantee_nothing":
+		return pdu.ReplicationGuaranteeKind_GuaranteeNothing, nil
+	case "guarantee_incremental":
+		return pdu.ReplicationGuaranteeKind_GuaranteeIncrementalReplication, nil
+	case "guarantee_resumability":
+		return pdu.ReplicationGuaranteeKind_GuaranteeResumability, nil
+	default:
+		return k, errors.Errorf("%q is not in guarantee_{nothing,incremental,resumability}", in)
+	}
+}

internal/replication/replication.go

@@ -0,0 +1,13 @@
+// Package replication implements replication of filesystems with existing
+// versions (snapshots) from a sender to a receiver.
+package replication
+
+import (
+	"context"
+
+	"github.com/zrepl/zrepl/internal/replication/driver"
+)
+
+func Do(ctx context.Context, driverConfig driver.Config, planner driver.Planner) (driver.ReportFunc, driver.WaitFunc) {
+	return driver.Do(ctx, driverConfig, planner)
+}

internal/replication/report/replication_report.go

@@ -0,0 +1,205 @@
+package report
+
+import (
+	"encoding/json"
+	"time"
+)
+
+type Report struct {
+	StartAt, FinishAt                      time.Time
+	WaitReconnectSince, WaitReconnectUntil time.Time
+	WaitReconnectError                     *TimedError
+	Attempts                               []*AttemptReport
+}
+
+var _, _ = json.Marshal(&Report{})
+
+type TimedError struct {
+	Err  string
+	Time time.Time
+}
+
+func NewTimedError(err string, t time.Time) *TimedError {
+	if err == "" {
+		panic("error must be empty")
+	}
+	if t.IsZero() {
+		panic("t must be non-zero")
+	}
+	return &TimedError{err, t}
+}
+
+func (s *TimedError) Error() string {
+	return s.Err
+}
+
+var _, _ = json.Marshal(&TimedError{})
+
+type AttemptReport struct {
+	State             AttemptState
+	StartAt, FinishAt time.Time
+	PlanError         *TimedError
+	Filesystems       []*FilesystemReport
+}
+
+type AttemptState string
+
+const (
+	AttemptPlanning      AttemptState = "planning"
+	AttemptPlanningError AttemptState = "planning-error"
+	AttemptFanOutFSs     AttemptState = "fan-out-filesystems"
+	AttemptFanOutError   AttemptState = "filesystem-error"
+	AttemptDone          AttemptState = "done"
+)
+
+type FilesystemState string
+
+const (
+	FilesystemPlanning        FilesystemState = "planning"
+	FilesystemPlanningErrored FilesystemState = "planning-error"
+	FilesystemStepping        FilesystemState = "stepping"
+	FilesystemSteppingErrored FilesystemState = "step-error"
+	FilesystemDone            FilesystemState = "done"
+)
+
+type FsBlockedOn string
+
+const (
+	FsBlockedOnNothing           FsBlockedOn = "nothing"
+	FsBlockedOnPlanningStepQueue FsBlockedOn = "plan-queue"
+	FsBlockedOnParentInitialRepl FsBlockedOn = "parent-initial-repl"
+	FsBlockedOnReplStepQueue     FsBlockedOn = "repl-queue"
+)
+
+type FilesystemReport struct {
+	Info *FilesystemInfo
+
+	State FilesystemState
+
+	// Always valid.
+	BlockedOn FsBlockedOn
+
+	// Valid in State = FilesystemPlanningErrored
+	PlanError *TimedError
+	// Valid in State = FilesystemSteppingErrored
+	StepError *TimedError
+
+	// Valid in State = FilesystemStepping
+	CurrentStep int
+	Steps       []*StepReport
+}
+
+type FilesystemInfo struct {
+	Name string
+}
+
+type StepReport struct {
+	Info *StepInfo
+}
+
+type StepInfo struct {
+	From, To        string
+	Resumed         bool
+	BytesExpected   uint64
+	BytesReplicated uint64
+}
+
+func (a *AttemptReport) BytesSum() (expected, replicated uint64, containsInvalidSizeEstimates bool) {
+	for _, fs := range a.Filesystems {
+		e, r, fsContainsInvalidEstimate := fs.BytesSum()
+		containsInvalidSizeEstimates = containsInvalidSizeEstimates || fsContainsInvalidEstimate
+		expected += e
+		replicated += r
+	}
+	return expected, replicated, containsInvalidSizeEstimates
+}
+
+func (f *FilesystemReport) BytesSum() (expected, replicated uint64, containsInvalidSizeEstimates bool) {
+	for _, step := range f.Steps {
+		expected += step.Info.BytesExpected
+		replicated += step.Info.BytesReplicated
+		containsInvalidSizeEstimates = containsInvalidSizeEstimates || step.Info.BytesExpected == 0
+	}
+	return
+}
+
+func (f *AttemptReport) FilesystemsByState() map[FilesystemState][]*FilesystemReport {
+	r := make(map[FilesystemState][]*FilesystemReport, 4)
+	for _, fs := range f.Filesystems {
+		l := r[fs.State]
+		l = append(l, fs)
+		r[fs.State] = l
+	}
+	return r
+}
+
+func (f *FilesystemReport) Error() *TimedError {
+	switch f.State {
+	case FilesystemPlanningErrored:
+		return f.PlanError
+	case FilesystemSteppingErrored:
+		return f.StepError
+	}
+	return nil
+}
+
+// may return nil
+func (f *FilesystemReport) NextStep() *StepReport {
+	switch f.State {
+	case FilesystemDone:
+		return nil
+	case FilesystemPlanningErrored:
+		return nil
+	case FilesystemSteppingErrored:
+		return nil
+	case FilesystemPlanning:
+		return nil
+	case FilesystemStepping:
+		// invariant is that this is always correct
+		// TODO what about 0-length Steps but short intermediary state?
+		return f.Steps[f.CurrentStep]
+	}
+	panic("unreachable")
+}
+
+func (f *StepReport) IsIncremental() bool {
+	return f.Info.From != ""
+}
+
+// Returns, for the latest replication attempt,
+// 0  if there have not been any replication attempts,
+// -1 if the replication failed while enumerating file systems
+// N  if N filesystems could not not be replicated successfully
+func (r *Report) GetFailedFilesystemsCountInLatestAttempt() int {
+
+	if len(r.Attempts) == 0 {
+		return 0
+	}
+
+	a := r.Attempts[len(r.Attempts)-1]
+	switch a.State {
+	case AttemptPlanningError:
+		return -1
+	case AttemptFanOutError:
+		var count int
+		for _, f := range a.Filesystems {
+			if f.Error() != nil {
+				count++
+			}
+		}
+		return count
+	default:
+		return 0
+	}
+}
+
+// Returns true in case the AttemptState is a terminal
+// state(AttemptPlanningError, AttemptFanOutError, AttemptDone)
+func (a AttemptState) IsTerminal() bool {
+	switch a {
+	case AttemptPlanningError, AttemptFanOutError, AttemptDone:
+		return true
+	default:
+		return false
+	}
+}