#!/bin/bash
# SPDX-License-Identifier: GPL-3.0-or-later
#
# bashclub zfs replication script
# Author: (C) 2024 Thorsten Spille <thorsten@spille-edv.de>

# version is taken from the installed package (set by the release tag)
__version__=v$(dpkg-query -W -f='${Version}' bashclub-zsync 2>/dev/null || echo dev)

# also search next to this script (manual installation, e.g. on a TrueNAS dataset)
PATH="$(dirname "$(readlink -f "$0")"):/bin:/sbin:/usr/bin:/usr/sbin:/usr/local/bin:/usr/local/sbin"

prog="$(basename $0)"
checkzfs=$(which checkzfs)
rsm=$(which remote-snapshot-manager)

rc=0
debug=

#### default config file, can be changed with parameter -c
conf=/etc/bashclub/zsync.conf

#### default values in config file
# replication target path on local machine 
target=pool/dataset

# ssh address of remote machine, "local" = datasets of this machine without ssh (pool to pool)
source=user@host

# ssh port of remote machine
sshport=22

# zfs user parameter to identify filesystems/volumes to replicate
tag=bashclub:zsync

# pipe separated list of the snapshot labels to replicate, each with the number of its snapshots to keep on this host:
#   daily or daily=source  the target follows the source (deletes what the source deleted, at least min_keep remain)
#   daily=30               the 30 newest snapshots of the label stay on this host, whatever the source deletes
snapshot_filter="hourly|daily|weekly|monthly"

# deprecated (will be removed in a later version): minimum number of snapshots per label that stay on this host for
# labels with "source"; use a number in snapshot_filter instead, e.g. daily=30
min_keep=3

# number of zfs snapshots to keep on source (0 or 1 = snapshot function disabled)
zfs_auto_snapshot_keep=0

# make snapshot via zfs-auto-snapshot before replication
zfs_auto_snapshot_label="backup"

# select snapshot engine: "zas" or "internal"
# keep "zas" here for existing configs without this setting; new configs get "internal"
zfs_auto_snapshot_engine="zas"

# snapshots of the label: "strict" = only the name scheme of the snapshot engine
# (internal: <label>-YYYY-MM-DD-HHMM, zas: zfs-auto-snap_<label>-YYYY-MM-DD-HHMM), "legacy" = any name containing the label
# keep "legacy" here for existing configs without this setting; new configs get "strict"
zfs_auto_snapshot_label_match="legacy"

# maximum age of the last successful run for the checkmk check bashclub_zsync, e.g. 26h, 90m or seconds
# empty = derived from the start times of the previous runs, 0 = no age check (e.g. removable disks)
monitor_max_age=""

# who starts the runs of this configuration: cron (also without this setting, as before), systemd (timer
# bashclub-zsync@<name>.timer, activate with: bashclub-zsync-timer sync), autorun (removable disk, zsync-autorun:
# no timer, no age check in the monitoring) or external (started by another tool, e.g. Miyagi)
trigger=""

# trigger=systemd: schedule in cron format, several separated by ";", e.g. "*/15 6-22 * * 1-5" or "@daily"
schedule=""

# trigger=systemd: user of the run (operation without root), empty = root
run_as=""

# trigger=systemd: append the output to this file, empty = journal (journalctl -u bashclub-zsync@<name>)
log_file=""

# maximum run time of a hook in seconds (hooks: /etc/bashclub/hooks.d, see README)
hook_timeout=600

# bandwidth limit of the transfer in bytes per second with suffix k, m, g (1024 based), e.g. 10m = 10 MiB/s
# (about 84 Mbit/s); applied on this host with pv (package pv), empty = no limit
bandwidth_limit=""

# keep a bookmark of the newest replicated snapshot on the source (one per backup host and target): if the rotation
# on the source deletes this snapshot before the next replication, the replication continues from the bookmark
# 1 = on, 0 = off; keep 0 here for existing configs without this setting, new configs get 1
source_bookmark=0

# hold the snapshots of a dataset on the source (zfs hold) while it is replicated, so the rotation or an admin cannot
# delete the base or a snapshot that is still to be sent; released at the end of the dataset or of the run
# 1 = on, 0 = off; keep 0 here for existing configs without this setting, new configs get 1
source_hold=0

# resumable transfers (zfs receive -s): a transfer interrupted by a network or power failure continues where it
# stopped in the next run instead of starting again; 1 = on, 0 = off
# keep 0 here for existing configs without this setting, new configs get 1
transfer_resume=0

# disable checkzfs with value > 0
checkzfs_disabled=0

# do not connect to source host (if replication is local)
checkzfs_local=0

# set checkzfs parameter "--prefix"
checkzfs_prefix=zsync

# set checkzfs maximum age of last snapshot in minutes (comma separated => warn,crit)
checkzfs_max_age=1500,6000

# set checkzfs maximum count of snapshots per dataset (comma separated => warn,crit)
checkzfs_max_snapshot_count=150,165

# set where to move the checkzfs output (0 = local machine, 1 = source machine)
checkzfs_spool=0

# set maxmimum age of checkzfs data in seconds (hourly = 3900, daily = 87000)
checkzfs_spool_maxage=87000

usage() {
	cat >&2 << EOF
$prog Version $__version__
-------------------------------------------------------------------------------
usage: $prog [-h] [-d] [-c CONFIG] [-s [checkmk]] [--result]
       $prog busy [--wait [SECONDS]]
    creates a mirrored replication of configured zfs filesystems / volumes
    -c CONFIG    Configuration file for this script
    -d           Debug mode
    -s           Status: run checkzfs, write the checkmk spool files and show the state, without replication
                 (without -c for all configurations in /etc/bashclub); -s checkmk shows the raw checkmk output
    --result     result of the last run of CONFIG as key=value (format=1), read from the run status, for scripts
    busy         is a replication, restore, maintenance action or removable disk run active? exit code 0 = nothing,
                 1 = active (listed); --wait waits until nothing is active (at most SECONDS, then exit code 1)
-------------------------------------------------------------------------------
    (C) 2024     by Spille IT Solutions for bashclub (github.com/bashclub)
                 Author: Thorsten Spille <thorsten@spille-edv.de>
-------------------------------------------------------------------------------
EOF
	exit $1
}

#### busy: is anything of bashclub-zsync active (for a tool that wants to shut the host down)
# replication or maintenance action: the lock of a configuration is held; restore job; removable disk (zsync-autorun,
# bashclub-zsync-disk); exit code 0 = nothing active, 1 = active
function busy_list() {
    local c pid phase f
    # runs of any configuration (also outside of /etc/bashclub): state file with a living process
    for f in /run/bashclub-zsync/*/*.state; do
        [[ -f $f ]] || continue
        pid=$(sed -n 's/^pid=//p' "$f"); c=$(sed -n 's/^config=//p' "$f")
        if [[ $pid =~ ^[0-9]+$ ]] && kill -0 "$pid" 2>/dev/null; then echo "replication: $c ($(sed -n 's/^phase=//p' "$f"))"; fi
    done
    # held lock of a configuration: a run without state (no status folder) or a maintenance action
    for c in /etc/bashclub/*.conf; do
        [[ -f $c ]] || continue
        if ! flock -n "$c" true 2>/dev/null; then echo "replication or maintenance: $c"; fi
    done
    for f in /run/bashclub-zsync-restore/*.state; do
        [[ -f $f ]] || continue
        pid=$(sed -n 's/^pid=//p' "$f"); phase=$(sed -n 's/^phase=//p' "$f")
        if [[ $phase =~ ^(planning|transfer|waiting|verify)$ ]] && [[ $pid =~ ^[0-9]+$ ]] && kill -0 "$pid" 2>/dev/null; then
            echo "restore job $(basename "$f" .state): $phase"
        fi
    done
    if command -v systemctl > /dev/null; then
        systemctl list-units --state=active --no-legend --plain 'zsync-autorun@*' 2>/dev/null | awk '{ print "removable disk: " $1 }'
    fi
    pgrep -f '^(/bin/bash |bash )?(/usr/bin/)?bashclub-zsync-disk create' > /dev/null && echo "removable disk: bashclub-zsync-disk create"
    return 0
}
if [[ $1 == "busy" ]]; then
    wait_s=
    if [[ $2 == "--wait" ]]; then wait_s=${3:-0}; [[ $wait_s =~ ^[0-9]+$ ]] || { echo "busy --wait: seconds expected" >&2; exit 2; }; fi
    start_s=$(date +%s)
    while true; do
        active=$(busy_list)
        if [[ -z $active ]]; then exit 0; fi
        if [[ -z $wait_s ]]; then echo "$active"; exit 1; fi
        if [ "$wait_s" -gt 0 ] && [ $(( $(date +%s) - start_s )) -ge "$wait_s" ]; then echo "$active"; exit 1; fi
        sleep 5
    done
fi

# --result (no short option): taken out before getopts
result_mode=0
filtered=()
for a in "$@"; do if [[ $a == "--result" ]]; then result_mode=1; else filtered+=("$a"); fi; done
set -- "${filtered[@]}"

args=("$@")
status_mode=0
conf_given=0
while getopts "hdc:s" opt; do
  case $opt in
    h) usage 0 ;;
    c) conf=$OPTARG; conf_given=1 ;;
    d) debug=-v ;;
    s) status_mode=1 ;;
    *) usage 1 ;;
  esac
done
shift $((OPTIND-1))
status_format=table
if [ $status_mode -eq 1 ] && [[ $1 == "checkmk" ]]; then status_format=checkmk; fi

# -s without -c: the status of all configurations, exit code of the worst one
if [ $status_mode -eq 1 ] && [ $conf_given -eq 0 ]; then
    worst=0
    for c in /etc/bashclub/*.conf; do
        [[ -f $c ]] || continue
        "$BASH" "$0" -c "$c" -s ${1:+"$1"}
        state=$?
        if [ $state -gt $worst ]; then worst=$state; fi
    done
    exit $worst
fi

function log() {
    # status mode (-s): only warnings and errors, on stderr (stdout is the state)
    if [ $status_mode -eq 1 ]; then
        if [[ $1 == *"[ERROR]"* || $1 == *"[WARN]"* ]]; then echo -e "$1" >&2; fi
        return
    fi
    echo -e "$(date +'%b %d %T') $1"
}

function debug() {
    if [[ $debug == "-v" ]]; then log "[DEBUG] $1"; fi
}

# bytes in readable form (10737418240 -> 10GiB)
function size() {
    numfmt --to=iec-i --suffix=B $1 2>/dev/null || echo "$1 bytes"
}

# stream between zfs send and zfs receive, limited to bandwidth_limit if pv is available
function throttle() {
    if [[ -n $throttle_rate ]]; then
        pv -q -L $throttle_rate
    else
        cat
    fi
}

# hold snapshots on the source (source_hold=1) until release_snaps; bookmarks and empty names are skipped
held_snaps=()
function hold_snaps() {
    local snaps=() s
    if [[ $source_hold != "1" ]]; then return 0; fi
    for s in "$@"; do
        if [[ $s == *@* ]]; then snaps+=("$s"); fi
    done
    if [ ${#snaps[@]} -eq 0 ]; then return 0; fi
    debug "Holding ${#snaps[@]} snapshots on the source ($hold_tag)"
    # a hold left over by an aborted run already exists: that is fine, the snapshot is held
    remote "for s in $(printf '%s ' "${snaps[@]}"); do zfs hold $hold_tag \$s 2>/dev/null; done; true"
    held_snaps+=("${snaps[@]}")
}
function release_snaps() {
    if [ ${#held_snaps[@]} -eq 0 ]; then return 0; fi
    debug "Releasing ${#held_snaps[@]} snapshots on the source ($hold_tag)"
    remote "for s in $(printf '%s ' "${held_snaps[@]}"); do zfs release $hold_tag \$s 2>/dev/null; done; true"
    held_snaps=()
}

# run a command on the source host (source=local: on this host without ssh)
function remote() {
    if [[ $source == "local" ]]; then
        bash -c "$*"
    else
        ssh $controlmaster $controlpath $controlpersist $sshcipher $zsync_sshport $source "$@"
    fi
}

#### run status for monitoring (checkmk local check bashclub_zsync), one directory per user, file names from the
# config path; the base directories have the sticky bit, each user writes only into its own directory
# /run/bashclub-zsync/<user>/<config>.state        active run: pid, phase, dataset (removed at the end of the run)
# /var/lib/bashclub-zsync/<user>/<config>.last     result of the last run and of the last successful run
# /var/lib/bashclub-zsync/<user>/<config>.history  start times of the last runs (interval for the age check)
# /var/lib/bashclub-zsync/<user>/<config>.skipped  times of runs skipped because of an active run
conf_path=$(readlink -m "$conf" 2>/dev/null || echo "$conf")
status_id=$(tr '/' '_' <<< "$conf_path")
status_file=
status_run_dir=/run/bashclub-zsync/$(id -un)
status_lib_dir=/var/lib/bashclub-zsync/$(id -un)
status_lib=$status_lib_dir/$status_id

#### --result: the last run of this configuration as key=value, from the status of the user that runs it (run_as,
# otherwise root, otherwise the newest); nothing is run, the source is not asked; unknown values are left out
if [ $result_mode -eq 1 ]; then
    owner=$(sed -n 's/^[[:space:]]*run_as=//p' "$conf_path" 2>/dev/null | tail -1 | tr -d "\"'" | cut -d' ' -f1)
    lib=
    d=/var/lib/bashclub-zsync/${owner:-root}
    if [[ -f $d/$status_id.last ]] && [[ $(stat -c %U "$d") == "$(basename "$d")" ]]; then lib=$d/$status_id; fi
    if [[ -z $lib ]]; then
        newest=0
        for f in /var/lib/bashclub-zsync/*/"$status_id.last"; do
            [[ -f $f ]] && [[ $(stat -c %U "$(dirname "$f")") == "$(basename "$(dirname "$f")")" ]] || continue
            if [ "$(stat -c %Y "$f")" -gt "$newest" ]; then newest=$(stat -c %Y "$f"); lib=${f%.last}; fi
        done
    fi
    echo "format=1"
    echo "config=$conf_path"
    running=0
    if [[ -f $conf_path ]] && ! flock -n "$conf_path" true 2>/dev/null; then running=1; fi
    echo "running=$running"
    if [[ -z $lib ]]; then echo "last_run=none"; exit 1; fi
    # result of the last run and of the last successful one (numbers only)
    while IFS='=' read -r key value; do
        case $key in
            start|end|duration|rc|datasets|datasets_failed|ok_start|ok_end|ok_duration)
                [[ $value =~ ^[0-9]*$ ]] && echo "$key=$value" ;;
        esac
    done < "$lib.last"
    # states of the datasets (see the Checkmk check) and the open decisions of bashclub-zsync-maintenance
    if [[ -f $lib.datasets ]]; then
        awk -F'\t' '$2 ~ /^[a-z_]+$/ { n[$2]++ } $2 ~ /^(orphan_missing|orphan_unselected|broken|empty|branch)$/ { cases++ }
            END { for (s in n) print "state." s "=" n[s]; print "decisions=" cases + 0 }' "$lib.datasets" | sort
    fi
    # transfers interrupted in the last run (receive_resume_token on the target): continued by the next run
    if [[ -f $lib.resume ]]; then
        echo "resume_datasets=$(grep -c . "$lib.resume")"
        while read -r ds; do [[ -n $ds ]] && echo "resume.$ds=1"; done < "$lib.resume"
    fi
    exit 0
fi

# own directory of this user: created if missing, used only if it is a directory of this user (not a link)
function own_dir() {
    mkdir -p -m 755 "$1" 2>/dev/null
    [ -d "$1" ] && [ ! -L "$1" ] && [ -O "$1" ] && [ -w "$1" ]
}

# phase of the active run: status <phase> [dataset]
function status() {
    if [[ -n $status_file ]]; then
        printf 'config=%s\npid=%s\npid_start=%s\nrun_start=%s\nphase=%s\ndataset=%s\nphase_since=%s\n' \
            "$conf_path" $$ "$pid_start" "$run_start" "$1" "$2" "$(date +%s)" > "$status_file.tmp" \
            && mv -f "$status_file.tmp" "$status_file"
    fi
}

# at the end of the run (trap EXIT): result of this run, the last successful run is kept if this one failed
function finish_status() {
    local code=${1:-$?} end=$(date +%s) key value
    local -A prev=()
    if [[ -f $status_lib.last ]]; then
        while IFS='=' read -r key value; do prev[$key]=$value; done < "$status_lib.last"
    fi
    if [ $code -eq 0 ]; then
        prev[ok_start]=$run_start; prev[ok_end]=$end; prev[ok_duration]=$((end - run_start))
    fi
    # marks=1: this version marks open decisions on the replicas (bashclub-zsync-maintenance)
    printf 'config=%s\nstart=%s\nend=%s\nduration=%s\nrc=%s\ndatasets=%s\ndatasets_failed=%s\nok_start=%s\nok_end=%s\nok_duration=%s\nmarks=1\n' \
        "$conf_path" $run_start $end $((end - run_start)) $code $datasets_total $datasets_failed \
        "${prev[ok_start]}" "${prev[ok_end]}" "${prev[ok_duration]}" > "$status_lib.last.tmp" \
        && mv -f "$status_lib.last.tmp" "$status_lib.last"
    # transfers left for the next run (receive_resume_token on the target), for --result
    if [[ $transfer_resume == "1" ]] && [[ -n $target ]]; then
        zfs get -H -o name,value -r receive_resume_token "$target" 2>/dev/null | awk -F'\t' -v t="$target/" '$2 != "-" && index($1, t) == 1 { print substr($1, length(t) + 1) }' \
            > "$status_lib.resume.tmp" && mv -f "$status_lib.resume.tmp" "$status_lib.resume"
    else
        rm -f "$status_lib.resume"
    fi
    rm -f "$status_file"
}

#### one run per config at a time: the script runs again under flock, a second run is skipped
# the lock is the configuration file itself: every user that runs it can read it, it lies in a directory only root
# writes (no lock file another user could replace), and root and run_as share it
# flock -o: the script itself does not hold the lock, so the background ssh master
# connection (ControlPersist) does not keep the lock after the run has finished
# only flock from util-linux (Linux): the options -o and -E are not available everywhere
# without the configuration (the initial config file is created below) there is nothing to lock
if [[ -z $ZSYNC_LOCK ]] && [[ -r $conf_path ]] && [ $status_mode -eq 0 ]; then
    if flock --version 2>/dev/null | grep -q util-linux; then
        export ZSYNC_LOCK=$conf_path
        flock -n -o -E 75 "$ZSYNC_LOCK" "$BASH" "$0" "${args[@]}"
        lock_rc=$?
        if [ $lock_rc -eq 75 ]; then
            active_pid=$(sed -n 's/^pid=//p' /run/bashclub-zsync/*/"$status_id.state" 2>/dev/null | head -1)
            log "[ERROR] Another run with configuration $conf is still active${active_pid:+ (pid $active_pid)}. Skipping this run."
            if own_dir "$status_lib_dir"; then
                { tail -n 19 "$status_lib.skipped" 2>/dev/null; date +%s; } > "$status_lib.skipped.tmp" \
                    && mv -f "$status_lib.skipped.tmp" "$status_lib.skipped"
            fi
            exit 1
        fi
        exit $lock_rc
    fi
    log "[WARN] flock (util-linux) not found, running without protection against concurrent runs."
fi

#### log file without logrotate (e.g. TrueNAS: cron job with ">> /mnt/<pool>/.../bashclub-zsync.log")
# if the output is appended to a file whose directory no logrotate config mentions,
# the script rotates it at 10 MiB and keeps 4 compressed old logs
logfile=$(readlink -f /proc/$$/fd/1 2>/dev/null)
if [[ -f $logfile ]] && [ $(stat -c %s "$logfile") -gt 10485760 ] \
    && ! cat /etc/logrotate.conf /etc/logrotate.d/* 2>/dev/null | grep -qF "$(dirname "$logfile")/"; then
    errfile=$(readlink -f /proc/$$/fd/2 2>/dev/null)
    for i in 3 2 1; do
        if [ -f "$logfile.$i.gz" ]; then mv -f "$logfile.$i.gz" "$logfile.$((i+1)).gz"; fi
    done
    mv -f "$logfile" "$logfile.1"
    exec >> "$logfile"
    if [[ $errfile == "$logfile" ]]; then exec 2>&1; fi
    gzip -f "$logfile.1"
    log "[INFO] Log file rotated, previous log: $logfile.1.gz"
fi

# load config file
if [ -f $conf ]; then
    log "[INFO] Reading configuration $conf"
    source $conf
else
    mkdir -p $(dirname $conf)
    cat << EOF > $conf
target=$target
source=$source
sshport=$sshport
tag=$tag
snapshot_filter="$snapshot_filter"
min_keep=$min_keep
zfs_auto_snapshot_keep=$zfs_auto_snapshot_keep
zfs_auto_snapshot_label=$zfs_auto_snapshot_label
zfs_auto_snapshot_engine=internal
zfs_auto_snapshot_label_match=strict
bandwidth_limit=$bandwidth_limit
source_bookmark=1
source_hold=1
transfer_resume=1
checkzfs_disabled=$checkzfs_disabled
checkzfs_local=$checkzfs_local
checkzfs_prefix=$checkzfs_prefix
checkzfs_max_age=$checkzfs_max_age
checkzfs_max_snapshot_count=$checkzfs_max_snapshot_count
checkzfs_spool=$checkzfs_spool
checkzfs_spool_maxage=$checkzfs_spool_maxage
trigger=$trigger
schedule="$schedule"
run_as=$run_as
log_file=$log_file
EOF
    log "[INFO] Initial config file created. Please adjust and restart script. Exiting..."
    usage 0
fi

# run status from here on (not for -h and not when the initial config file was created)
run_start=$(date +%s)
datasets_total=0
datasets_failed=0
# own directories of this user below the base directories (the package lets the group bashclub-zsync create them,
# root creates the base directories on a manual installation)
if [ "$(id -u)" -eq 0 ]; then mkdir -p /run/bashclub-zsync /var/lib/bashclub-zsync 2>/dev/null; fi
if [ $status_mode -eq 1 ]; then
    : # status mode: no run, the run status stays as it is
elif own_dir "$status_run_dir" && own_dir "$status_lib_dir"; then
    status_file=$status_run_dir/$status_id.state
    # start time of the process, to tell a reused pid apart
    pid_start=$(sed 's/.*) //' /proc/$$/stat 2>/dev/null | cut -d' ' -f20)
    { tail -n 10 "$status_lib.history" 2>/dev/null; echo $run_start; } > "$status_lib.history.tmp" \
        && mv -f "$status_lib.history.tmp" "$status_lib.history"
    status start
else
    log "[WARN] $status_run_dir or $status_lib_dir cannot be used by $(id -un) (missing, of another user, or $(id -un) is not in the group bashclub-zsync), no run status for the monitoring."
fi

#### checkmk spool (/var/lib/check_mk_agent/spool), local or on the source (checkzfs_spool), atomically: a new file of
# its own in the spool (mktemp) is renamed, the agent never reads a partial file and nothing follows a link
spooldir=/var/lib/check_mk_agent/spool
source_reachable=0
# the spool on the source: checkzfs_spool=1, not checkzfs_local, and the source answered in this run
function spool_on_source() {
    [ $checkzfs_spool -ne 0 ] && [ $checkzfs_local -eq 0 ] && [ $source_reachable -eq 1 ]
}
# write_spool <spool name> <file with the content>; the file is removed
function write_spool() {
    local spoolname=$1 file=$2 spooltmp
    chmod 644 "$file"
    if ! spool_on_source; then
        mkdir -p $spooldir 2>/dev/null
        if [ -w $spooldir ]; then
            spooltmp=$(mktemp $spooldir/.$spoolname.XXXXXX) && cat "$file" > $spooltmp && chmod 644 $spooltmp \
                && mv -f $spooltmp $spooldir/$spoolname
        else
            log "[WARN] $spooldir is not writable for $(id -un), $spoolname is not stored."
        fi
    else
        # over the ssh connection of the run (no scp: works with the forced command bashclub-zsync-remote)
        spooltmp=$spooldir/$(basename "$file")
        remote "mkdir -p $spooldir"
        remote "cat > $spooltmp" < "$file" && remote "mv $spooltmp $spooldir/$spoolname"
    fi
    rm -f "$file"
}

#### monitoring: one spool file per run, <checkzfs_spool_maxage>_<checkzfs_prefix> as checkzfs wrote it before 2.0 (the
# number is the maximum age the agent accepts), with the checkzfs services and the state of this configuration as
# service "<checkzfs_prefix> <config>" (the evaluation of the check bashclub_zsync); written at the end of each run,
# during a run (watchdog) and with -s, locally or on the source (checkzfs_spool); with the spool on the source the
# services stay visible while the backup server is switched off (Miyagi workflow), until the file is older than its
# maxage. Several backup servers on one source need different checkzfs_prefix, as for checkzfs before.
status_check=/usr/share/bashclub-zsync/checkmk/bashclub_zsync
monitor_spool=${checkzfs_spool_maxage}_${checkzfs_prefix}
# checkzfs output of the last run (the watchdog writes it again with the current state)
checkzfs_lines=
checkzfs_cache=$status_lib.checkzfs
function status_lines() {
    [[ -x $status_check ]] && [[ ! -e /etc/bashclub/zsync-checkmk.disabled ]] || return 0
    if [[ $checkzfs_prefix != "zsync" ]] && [[ $checkzfs_prefix =~ ^[A-Za-z0-9_.:-]+$ ]]; then
        "$status_check" --config "$conf_path" 2>/dev/null | sed -E "s/^([0-3P]) \"zsync /\1 \"$checkzfs_prefix /"
    else
        "$status_check" --config "$conf_path" 2>/dev/null
    fi
}
function write_monitor_spool() {
    local file lines=0
    file=$(mktemp /tmp/.$monitor_spool.XXXXXX) || return 0
    {
        echo "<<<local>>>"
        if [ $checkzfs_disabled -eq 0 ]; then
            if [[ -n $checkzfs_lines ]]; then echo "$checkzfs_lines"
            elif [[ -f $checkzfs_cache ]]; then grep -v '^<<<' "$checkzfs_cache"; fi
        fi
        status_lines
    } > "$file"
    lines=$(grep -vc '^<<<' "$file")
    if [ "$lines" -eq 0 ]; then rm -f "$file"; return 0; fi
    write_spool "$monitor_spool" "$file"
    # testing versions of 2.0 wrote the state into files of their own: removed once (on the source with an older
    # bashclub-zsync-remote they stay until their maxage, the agent ignores them then)
    if [[ -n $status_file ]] && [[ ! -e $status_lib.spool-merged ]]; then
        local old name
        name=$(basename "$conf" .conf | tr -c 'A-Za-z0-9_.\n-' '_')
        for old in "${checkzfs_spool_maxage}_bashclub-zsync-$name" \
            "${checkzfs_spool_maxage}_bashclub-zsync-$(hostname -s | tr -c 'A-Za-z0-9_.\n-' '_')-$name"; do
            if spool_on_source; then remote "rm -f $spooldir/$old" > /dev/null 2>&1; fi
            [[ -w $spooldir ]] && rm -f "$spooldir/$old"
        done
        touch "$status_lib.spool-merged" 2>/dev/null
    fi
}
# checkzfs at the end of a run and with -s; run_checkzfs [file]: with a file name the output is also copied there
function run_checkzfs() {
    local c_key= c_value=
    if [[ $checkzfs_local -eq 0 ]]; then
        c_key=--source
        c_value=$source:$sshport
    fi
    debug "Command:\n$checkzfs $c_key $c_value --output checkmk --threshold $checkzfs_max_age --maxsnapshots $checkzfs_max_snapshot_count --prefix $checkzfs_prefix --replicafilter "^${target}" --filter \"$checkzfs_filter\""
    checkzfs_lines=$($checkzfs $c_key $c_value --output checkmk --threshold $checkzfs_max_age --maxsnapshots $checkzfs_max_snapshot_count --prefix $checkzfs_prefix --replicafilter "^${target}" --filter "$checkzfs_filter" | grep -v '^<<<')
    if [[ -n $1 ]]; then printf '%s\n' "$checkzfs_lines" > "$1"; fi
    if [[ -n $checkzfs_lines ]] && [[ -d $status_lib_dir ]] && [[ -O $status_lib_dir ]]; then
        printf '%s\n' "$checkzfs_lines" > "$checkzfs_cache.tmp" && mv -f "$checkzfs_cache.tmp" "$checkzfs_cache"
    fi
}
# during a run: every ZSYNC_WATCHDOG_INTERVAL seconds (default 5 min), on the source only if the state changed
watchdog_pid=
function start_watchdog() {
    [[ -x $status_check ]] && [[ ! -e /etc/bashclub/zsync-checkmk.disabled ]] || return 0
    (
        trap - EXIT
        written=
        while sleep "${ZSYNC_WATCHDOG_INTERVAL:-300}"; do
            level=$(status_lines | head -1 | cut -d' ' -f1)
            if [ $checkzfs_spool -eq 0 ] || [ $checkzfs_local -gt 0 ] || [[ $level != "$written" ]]; then
                write_monitor_spool > /dev/null 2>&1
                written=$level
            fi
        done
    ) &
    watchdog_pid=$!
}

# end of the run (also on errors): release the holds on the source, write the result for the monitoring
function on_exit() {
    local code=$?
    if [[ -n $watchdog_pid ]]; then kill $watchdog_pid 2>/dev/null; fi
    release_snaps
    if [[ -n $status_file ]]; then finish_status $code; fi
    # hooks after the run, also after an error or a skipped run (e.g. to undo what pre-run did); exit code unchanged
    if [ $hooks_started -eq 1 ]; then
        hook_rc=$code
        run_hooks post-run || log "[WARN] a post-run hook failed, the exit code of the run stays $code."
    fi
    # status mode (-s) wrote it itself
    if [ $status_mode -eq 0 ]; then write_monitor_spool; fi
    if [[ -n $inhibit_pid ]]; then kill $inhibit_pid 2>/dev/null; fi
}
#### hooks: executable files in /etc/bashclub/hooks.d/<pre-run|post-run>/ (all configurations) and in
# /etc/bashclub/hooks.d/<config>/<pre-run|post-run>/, in lexical order, with the environment ZSYNC_CONFIG, ZSYNC_NAME,
# ZSYNC_TARGET, ZSYNC_SOURCE, ZSYNC_PHASE and after the run ZSYNC_RC and ZSYNC_DURATION; only files and folders of root
# that only root may write (like the configuration), their output goes into the log of the run
hooks_dir=${ZSYNC_HOOKS_DIR:-/etc/bashclub/hooks.d}
hooks_started=0
hook_rc=
function root_only() {
    [[ ! -L $1 ]] && [[ $(stat -c %u "$1" 2>/dev/null) == "${ZSYNC_CONF_OWNER:-0}" ]] \
        && [ $(( 8#$(stat -c %a "$1") & 8#022 )) -eq 0 ]
}
# run_hooks <pre-run|post-run>: 1 if a hook failed or cannot be trusted (the hooks after it are not run)
function run_hooks() {
    local phase=$1 dir hook hrc timeout_s=$hook_timeout name
    name=$(basename "$conf" .conf)
    [[ $timeout_s =~ ^[0-9]+$ ]] && [ $timeout_s -gt 0 ] || timeout_s=600
    for dir in "$hooks_dir/$phase" "$hooks_dir/$name/$phase"; do
        [[ -e $dir ]] || continue
        if ! [[ -d $dir ]] || ! root_only "$hooks_dir" || ! root_only "$dir" || { [[ $dir == "$hooks_dir/$name/"* ]] && ! root_only "$hooks_dir/$name"; }; then
            log "[ERROR] $dir is not a folder of root that only root may write, its hooks are not run."
            return 1
        fi
        for hook in "$dir"/*; do
            [[ -e $hook ]] || continue
            if ! [[ -f $hook && -x $hook ]] || ! root_only "$hook"; then
                log "[ERROR] $hook is not an executable file of root that only root may write, it is not run."
                return 1
            fi
            log "[INFO] $phase hook $hook"
            ZSYNC_CONFIG=$conf_path ZSYNC_NAME=$name ZSYNC_TARGET=$target ZSYNC_SOURCE=$source ZSYNC_PHASE=$phase \
                ZSYNC_RC=$hook_rc ZSYNC_DURATION=$(( $(date +%s) - run_start )) \
                timeout "$timeout_s" "$hook" < /dev/null 2>&1 | while IFS= read -r line; do log "[INFO] ${hook##*/}: $line"; done
            hrc=${PIPESTATUS[0]}
            if [ $hrc -ne 0 ]; then
                log "[$([[ $phase == pre-run ]] && echo ERROR || echo WARN)] $phase hook $hook failed (exit code $hrc$([ $hrc -eq 124 ] && echo ", stopped after ${timeout_s}s"))."
                return 1
            fi
        done
    done
    return 0
}
trap on_exit EXIT
# no shutdown or sleep in the middle of a run (systemd-inhibit; root can override it, a tool asks "bashclub-zsync busy");
# a user without the polkit right gets none, the run goes on
inhibit_pid=
if [ $status_mode -eq 0 ] && [ -d /run/systemd/system ] && command -v systemd-inhibit > /dev/null; then
    systemd-inhibit --what=shutdown:sleep --mode=block --who=bashclub-zsync --why="replication $conf_path" tail --pid=$$ -f /dev/null > /dev/null 2>&1 &
    inhibit_pid=$!
fi

# checked after loading the config, so checkzfs_disabled from the config is respected
if [[ -n $checkzfs ]]; then
    if [ $checkzfs_disabled -eq 0 ]; then
        log "[INFO] installed checkzfs found: $checkzfs"
    else
        log "[INFO] checkzfs is disabled by configuration."
    fi
else
    checkzfs_disabled=1
    log "[WARN] checkzfs is disabled. for monitoring please install:\n\thttps://code.bashclub.org/bashclub/checkzfs\n"
fi

if [[ -z $source ]]; then
    log "[ERROR] source is empty. For local replication without ssh use source=local"
    exit 1
fi
zsync_sshport=-p$sshport

config_log="[INFO] Configuration:"
for setting in target source sshport tag snapshot_filter min_keep zfs_auto_snapshot_keep zfs_auto_snapshot_label \
               zfs_auto_snapshot_engine zfs_auto_snapshot_label_match \
               bandwidth_limit source_bookmark source_hold transfer_resume checkzfs_disabled checkzfs_local checkzfs_prefix checkzfs_max_age checkzfs_max_snapshot_count \
               checkzfs_spool checkzfs_spool_maxage trigger schedule run_as log_file; do
    config_log+="\n\t$setting=${!setting}"
done
log "$config_log\n"

#### hooks before the run (inside the lock of the run): a failing one skips the run
if [ $status_mode -eq 0 ]; then
    status hooks
    hooks_started=1
    if ! run_hooks pre-run; then
        log "[ERROR] a pre-run hook failed, the run is skipped."
        exit 1
    fi
fi

status prepare

#### ssh connection: shared master connection, aes cipher if both hosts support it
controlmaster=-oControlMaster=auto
# socket: %C (hash of host, port and user, 40 characters) keeps the path below the limit of 107 characters;
# with a long home directory (e.g. users of bashclub-zsync-setup) in a private directory in /tmp (sticky, owner
# checked), without it the connections are not shared
controlpath=-oControlPath=~/.ssh/zsync-%C
if [ ${#HOME} -gt 38 ]; then
    ctl_dir=/tmp/bashclub-zsync-$(id -u)
    mkdir -m 700 "$ctl_dir" 2> /dev/null
    if [ -d "$ctl_dir" ] && [ ! -L "$ctl_dir" ] && [ -O "$ctl_dir" ] && [[ $(stat -c %a "$ctl_dir" 2> /dev/null) == "700" ]]; then
        controlpath=-oControlPath=$ctl_dir/%C
    else
        controlmaster=-oControlMaster=no
    fi
fi
controlpersist=-oControlPersist=15

if [[ $source == "local" ]]; then
    log "[INFO] Local mode: replicating datasets of this host without ssh (source=local)"
    # checkzfs checks the local pools, its output goes to the local spool
    checkzfs_local=1
fi

local_os_id=$(grep -E "^ID=" /etc/os-release | cut -d'=' -f2)
# cipher: aes256-gcm if both hosts support aes (not needed in local mode)
if [[ $source != "local" ]]; then
    remote_os_id=$(remote "grep -E \"^ID=\" /etc/os-release | cut -d'=' -f2")

    if [[ $local_os_id == "freebsd" ]]; then
        local_aes=$(grep -o AES /var/run/dmesg.boot | uniq)
    else
        local_aes=$(grep -m1 -o aes /proc/cpuinfo | uniq)
    fi

    if [[ $remote_os_id == "freebsd" ]]; then
        remote_aes=$(remote "grep -o AES /var/run/dmesg.boot | uniq")
    else
        remote_aes=$(remote "grep -m1 -o aes /proc/cpuinfo | uniq")
    fi

    if [[ $local_aes == "aes" ]] && [[ $remote_aes == "aes" ]]; then
        debug "Switching to cipher aes256-gcm@openssh.com"
        sshcipher=-caes256-gcm@openssh.com
    else
        debug "Using default cipher chacha20-poly1305@openssh.com"
        sshcipher=-cchacha20-poly1305@openssh.com
    fi
fi

#### datasets to replicate: zfs user property $tag on the source
# all     = this dataset (if set locally or inherited, not received)
# subvols = only child datasets (the dataset with the local setting itself is skipped)
include_list="[INFO] Included datasets:\n"
exclude_list="[INFO] Excluded datasets:\n"
checkzfs_headline="[INFO] checkzfs filter:"
checkzfs_filter=
# critical: without the list of the source nothing can be replicated (ssh failed or zfs failed on the source)
tagged=$(remote "zfs get -H -o name,value,source -t filesystem,volume $tag")
remote_rc=$?
if [ $remote_rc -ne 0 ]; then
    if [[ $source == "local" ]]; then
        log "[ERROR] zfs get $tag failed (exit code $remote_rc), nothing is replicated."
    else
        log "[ERROR] source $source (port $sshport) is not reachable or zfs failed there (exit code $remote_rc), nothing is replicated."
    fi
    exit 1
fi
source_reachable=1
# watchdog from here on: it is a subshell and knows whether the spool goes to the source only from this point
if [[ -n $status_file ]] && [ $status_mode -eq 0 ]; then start_watchdog; fi
IFS=$'\n'
for zvol in $tagged; do
    # property source: local, received, inherited from <dataset> or default
    IFS=$'\t' read -r name value property_source <<< "$zvol"
    # local mode: the replicas in the target can inherit the property from a dataset of the same pool
    if [[ $source == "local" ]] && { [[ $name == "$target" ]] || [[ $name == "$target/"* ]]; }; then
        continue
    fi
    if [[ $value == "subvols" ]]; then
        if [[ $property_source != "local" ]] && [[ $property_source != "received" ]]; then
            include_list="${include_list}\t${name}\n"
            syncvols+=("$name")
        elif [[ $property_source == "local" ]]; then
            checkzfs_filter="#$name/|$checkzfs_filter"
        fi
    elif [[ $value == "all" ]]; then
        if [[ $property_source != "received" ]]; then
            include_list="${include_list}\t${name}\n"
            syncvols+=("$name")
        fi
        if [[ $property_source == "local" ]]; then
            checkzfs_filter="#$name|$checkzfs_filter"
        fi
    else
        exclude_list="${exclude_list}\t${name}\n"
    fi
done
checkzfs_filter=${checkzfs_filter::-1}
# type and refquota (bytes, 0 = none, "-" for volumes) of all datasets on the source
declare -A fstypes refquotas
while IFS=$'\t' read -r dataset fstype refquota; do
    fstypes[$dataset]=$fstype
    refquotas[$dataset]=$refquota
done <<< "$(remote "zfs list -Hp -o name,type,refquota -t filesystem,volume")"
# original values of the properties the replication changes or leaves out (mountpoint, canmount, auto-snapshot,
# refquota, tag) and where they come from, saved on the replica for a restore (save_source_props):
# bashclub:zsync-source.<property>=<local|inherited|default|received|->:<value>
declare -A source_props
while IFS=$'\t' read -r dataset property value property_source; do
    if [[ -z $dataset ]]; then continue; fi
    if [[ $property_source == inherited* ]]; then property_source=inherited; fi
    source_props[$dataset]+="bashclub:zsync-source.$property=$property_source:$value"$'\n'
done <<< "$(remote "zfs get -H -o name,property,value,source -t filesystem,volume mountpoint,canmount,com.sun:auto-snapshot,refquota,$tag 2>/dev/null")"
log "$include_list"
log "$exclude_list"
log "$checkzfs_headline\ncheckzfs_filter=$checkzfs_filter\n"

#### status mode (-s): checkzfs and the state of this configuration into the spool and onto the terminal, nothing else
# (no snapshot, replication, deletion or change of the run status); exit code like checkmk: 0 OK, 1 WARN, 2 CRIT, 3 UNKNOWN
if [ $status_mode -eq 1 ]; then
    shown=$(mktemp)
    if [ $checkzfs_disabled -eq 0 ]; then
        run_checkzfs "$shown"
    else
        echo "checkzfs is disabled or not installed, no checkzfs services" >&2
    fi
    status_now=$(status_lines)
    write_monitor_spool
    printf '%s\n' "$status_now" >> "$shown"
    if [[ -n $(sed -n 's/^pid=//p' "$status_run_dir/$status_id.state" /run/bashclub-zsync/*/"$status_id.state" 2>/dev/null) ]]; then
        echo "a run of $conf is active, it writes the spool again at its end" >&2
    fi
    worst=0
    while IFS= read -r line; do
        [[ $line =~ ^[0-3P]\  ]] || continue
        state=${line%% *}; [[ $state == P ]] && state=0
        if [ $state -gt $worst ]; then worst=$state; fi
        if [[ $status_format == "checkmk" ]]; then echo "$line"; continue; fi
        rest=${line#* }
        if [[ $rest == \"* ]]; then service=${rest#\"}; service=${service%%\"*}; rest=${rest#\"$service\" }
        else service=${rest%% *}; rest=${rest#* }; fi
        text=${rest#* }
        printf '%-5s %-40s %s\n' "$(case $state in 0) echo OK ;; 1) echo WARN ;; 2) echo CRIT ;; *) echo UNKN ;; esac)" "$service" "$text"
    done < "$shown"
    rm -f "$shown"
    exit $worst
fi

#### target dataset, never snapshotted by zfs-auto-snapshot on this host
if ! zfs list $target > /dev/null 2>&1 ; then
    debug "$target does not exist. Creating..."
    zfs create -o canmount=noauto -o com.sun:auto-snapshot=false $target
else
    debug "$target exists, check auto-snapshot..."
    if [[ $(zfs get -H -o value,source com.sun:auto-snapshot $target) != "false	local" ]]; then
        zfs set com.sun:auto-snapshot=false $target
    fi
fi

#### snapshots to keep per label (snapshot_filter="hourly|daily=30"): label=N keeps the N newest snapshots of the label on
# this host, independent of the source; label, label=source and label=0 follow the source (with min_keep, as before)
# snapshot_filter keeps only the labels from here on
declare -A target_keep=()
filter_labels=
IFS='|' read -ra filter_entries <<< "$snapshot_filter"
for entry in "${filter_entries[@]}"; do
    filter_label=${entry%%=*}
    if [[ -z $filter_label ]]; then continue; fi
    if [[ $entry == *=* ]]; then
        keep_value=${entry#*=}
        if [[ $keep_value =~ ^[0-9]+$ ]] && [ $keep_value -gt 0 ]; then
            target_keep[$filter_label]=$keep_value
        elif [[ $keep_value != "source" ]] && [[ $keep_value != "0" ]]; then
            log "[WARN] Invalid number of snapshots to keep in snapshot_filter: $entry (a number or source), the target follows the source for $filter_label."
        fi
    fi
    filter_labels+="${filter_labels:+|}$filter_label"
done
snapshot_filter=$filter_labels

#### snapshots on the source before replication (zas: once for all datasets, internal: per dataset below)
if [ $zfs_auto_snapshot_keep -gt 1 ]; then
    if [[ $zfs_auto_snapshot_engine == "zas" ]]; then
        # without zfs-auto-snapshot the call below creates no snapshots and ends without a message
        # (exit code 1 = not found; 126 = the forced command of an earlier version does not know the check)
        remote "which zfs-auto-snapshot" > /dev/null 2>&1
        if [ $? -eq 1 ]; then
            log "[WARN] zfs-auto-snapshot is not installed on the source, the snapshot engine zas creates no snapshots. Install it there (apt install zfs-auto-snapshot) or use zfs_auto_snapshot_engine=internal."
        fi
        debug "Running zfs-auto-snapshot"
        remote "which zfs-auto-snapshot > /dev/null || exit 0 ; zfs-auto-snapshot --quiet --syslog --label=$zfs_auto_snapshot_label --keep=$zfs_auto_snapshot_keep //"
    fi

    if [[ $snapshot_filter == "" ]]; then
        snapshot_filter="$zfs_auto_snapshot_label"
    elif [[ "|$snapshot_filter|" != *"|$zfs_auto_snapshot_label|"* ]]; then
        # only if not already in the filter, otherwise each interval is checked twice
        snapshot_filter="$snapshot_filter|$zfs_auto_snapshot_label"
    fi
fi

#### snapshot name patterns (grep -E): all snapshots to replicate and the snapshots of each interval
# legacy: "@.*<interval>" for every interval including the label (unchanged behavior)
# strict: the label only matches the name scheme of the snapshot engine, so snapshots left over from
#         another engine (e.g. zfs-auto-snap_<label>-... after switching to internal) or from a label
#         that contains this label (e.g. <label>-old-...) are neither replicated nor used for deletion
if [[ $zfs_auto_snapshot_label_match != "strict" ]] && [[ $zfs_auto_snapshot_label_match != "legacy" ]]; then
    log "[WARN] Unknown value zfs_auto_snapshot_label_match=$zfs_auto_snapshot_label_match, using legacy."
    zfs_auto_snapshot_label_match=legacy
fi
#### bookmark and hold tag on the source per backup host and target, e.g. pool/data#zsync_backup1_backup_pve1
bookmark_name="zsync_$(hostname -s)_$(tr '/' '_' <<< "$target")"
hold_tag=$bookmark_name

#### without root (zfs allow): receive without mounting, on Linux only root may mount
nomount=
if [ "$(id -u)" -ne 0 ]; then nomount=-u; fi

# without root on a target with mountpoint=none (bashclub-zsync-setup, README): no replica is ever mounted there and
# the user may not set canmount (it could make root mount a replica), canmount is left as it is
function unmounted_target() {
    [ -n "$nomount" ] && [[ $(zfs get -H -o value mountpoint "$1" 2>/dev/null) == "none" ]]
}

#### resumable transfers: receive -s keeps the partial state of an interrupted transfer (resume token)
resumable=
if [[ $transfer_resume == "1" ]]; then resumable=-s; fi

#### bandwidth limit of the transfer: pv between send and receive on this host, without pv no limit
throttle_rate=
if [[ -n $bandwidth_limit ]]; then
    if ! [[ $bandwidth_limit =~ ^[0-9]+[kKmMgGtT]?$ ]]; then
        log "[WARN] Invalid value bandwidth_limit=$bandwidth_limit (e.g. 10m), running without bandwidth limit."
    elif ! which pv > /dev/null; then
        log "[WARN] pv not found, running without bandwidth limit. Install pv to limit the bandwidth."
    else
        throttle_rate=$bandwidth_limit
        log "[INFO] Bandwidth limit: $bandwidth_limit bytes per second (pv)"
    fi
fi

if [[ $zfs_auto_snapshot_engine == "zas" ]]; then
    label_scheme="zfs-auto-snap_$zfs_auto_snapshot_label"
else
    label_scheme="$zfs_auto_snapshot_label"
fi
# name ends after the date, followed by a tab (lines with guid) or the end of the line
label_regex="@$label_scheme-[0-9]{4}-[0-9]{2}-[0-9]{2}-[0-9]{4}([[:space:]]|\$)"
declare -A interval_regex
for interval in $(sed "s/|/\n/g" <<< "$snapshot_filter"); do
    interval_regex[$interval]="@.*$interval"
done
if [ $zfs_auto_snapshot_keep -gt 1 ] && [[ $zfs_auto_snapshot_label_match == "strict" ]]; then
    interval_regex[$zfs_auto_snapshot_label]=$label_regex
    snapshot_regex=$(IFS='|'; echo "(${interval_regex[*]})")
else
    snapshot_regex="@.*($snapshot_filter)"
fi

# snapshots whose name contains the label but does not match the name scheme of the snapshot engine:
# not created by this configuration, but in legacy mode replicated and used for deletion
# save the original values of the source on the replica (only those that changed since the last run)
function save_source_props() {
    local name=$1 replica=$target/$1 saved want changed=()
    if [[ -z ${source_props[$name]} ]] || ! zfs list "$replica" > /dev/null 2>&1; then return 0; fi
    saved=$(zfs get -H -o property,value -s local all "$replica" 2>/dev/null | grep '^bashclub:zsync-source\.' | tr '\t' '=')
    while read -r want; do
        if [[ -n $want ]] && ! grep -qxF "$want" <<< "$saved"; then changed+=("$want"); fi
    done <<< "${source_props[$name]}"
    if [ ${#changed[@]} -gt 0 ]; then
        debug "$name - saving the original properties on $replica: $(printf '%s ' "${changed[@]}")"
        zfs set "${changed[@]}" "$replica" \
            || log "[WARN] $name - saving the original properties of the source on $replica failed (bashclub:zsync-source.*)."
    fi
}

function label_leftovers() {
    grep -E "@.*$zfs_auto_snapshot_label" | grep -vE "$label_regex"
}

# state of each dataset for the monitoring (written to <config>.datasets):
# ok, branched, cleanup_blocked, failed, at_risk, diverged, broken, empty, no_snapshots, orphan_missing, orphan_unselected,
# orphan_kept, branch (old chain after a rollback on the source), branch_kept
declare -A ds_state ds_detail ds_leftovers

# fix for a target without a usable snapshot: keep the old replica under a new name, the next run replicates again
# newest target snapshot missing on the source (rollback there, or its newest snapshots deleted): the replication
# continues on a new chain from the newest common snapshot, the old chain stays complete as <replica>.zsync-branch-<date>
# (bashclub:zsync-branch=<date>;<reason>); nothing is deleted, the admin decides with bashclub-zsync-maintenance
# clone of the common snapshot -> promote (the common snapshots belong to the new chain) -> swap the names -> children
# (replicas of their own) back below the new chain
function branch_replica() {
    local name=$1 common=$2 replica=$target/$1 date new branch discard count prop value child
    date=$(date +%Y-%m-%d-%H%M)
    new=$replica.zsync-new
    branch=$replica.zsync-branch-$date
    if zfs list "$new" > /dev/null 2>&1 || zfs list "$branch" > /dev/null 2>&1; then
        log "[ERROR] $name - $new or $branch exists already (an earlier attempt was not completed), check them and remove what is not needed."
        return 1
    fi
    discard=$(zfs list -H -o name -S createtxg -t snapshot "$replica" | awk -v c="$common" '$1 == c { exit } { print }')
    count=$(grep -c . <<< "$discard")
    log "[WARN] $name - the newest snapshots of $replica do not exist on the source anymore (rollback or deleted there). The replication continues on a new chain from $common, the old chain with all its snapshots stays as $branch ($count newer snapshots):\n$(sed 's/^/\t/' <<< "$discard")"
    # as non-root zfs clone cannot mount it and fails, the clone exists nevertheless
    if [[ $fstype == "filesystem" ]] && [ -z "$nomount" ]; then
        zfs clone -o canmount=noauto "$common" "$new"
    else
        zfs clone "$common" "$new" 2> /dev/null
    fi
    if ! zfs list "$new" > /dev/null 2>&1; then
        log "[ERROR] $name - cloning $common failed, nothing was changed."
        return 1
    fi
    # properties of the replica (received or set by bashclub-zsync, e.g. bashclub:zsync-source.*) also on the new chain,
    # those that cannot be set there (encryption, missing permission) are left out
    while IFS=$'\t' read -r prop value; do
        if [[ -n $prop ]]; then zfs set "$prop=$value" "$new" 2> /dev/null; fi
    done <<< "$(zfs get -H -o property,value -s local,received all "$replica" 2>/dev/null)"
    if ! zfs promote "$new"; then
        log "[ERROR] $name - promoting $new failed, nothing was changed."
        zfs destroy "$new"
        return 1
    fi
    if ! zfs rename "$replica" "$branch" || ! zfs rename "$new" "$replica"; then
        log "[ERROR] $name - renaming failed: the new chain is $new, the old chain $replica or $branch. Rename them by hand: zfs rename $replica $branch; zfs rename $new $replica"
        return 1
    fi
    for child in $(zfs list -H -o name -d 1 -t filesystem,volume "$branch" | tail -n +2); do
        zfs rename "$child" "$replica/${child##*/}" || log "[ERROR] $name - moving $child below the new chain failed: zfs rename $child $replica/${child##*/}"
    done
    zfs set "bashclub:zsync-branch=$date;newest common snapshot ${common#*@}, $count newer snapshots" "$branch" \
        || log "[WARN] $name - marking $branch failed (bashclub:zsync-branch)."
    ds_state[$name]=branched
    ds_detail[$name]="new chain from ${common#*@}, old chain ${branch##*/}"
    return 0
}

function rename_hint() {
    echo "zfs rename $target/$1 $target/$1.broken-$(date +%Y-%m-%d)"
}

#### replication per dataset
for name in "${syncvols[@]}"; do
    log "[INFO] Replicate $name"
    status replicate "$name"
    # failed datasets are counted for the monitoring, rc of the whole run stays 1 after a failure
    ((datasets_total++))
    rc_before=$rc
    rc=0
    fstype=${fstypes[$name]}
    if [[ $fstype == "filesystem" ]]; then
        mp=-xmountpoint
        cm=-ocanmount=noauto
    else
        mp=
        cm=
    fi

    if [ $zfs_auto_snapshot_keep -gt 1 ]; then
        if [[ $zfs_auto_snapshot_engine == "internal" ]]; then
            debug "Running remote-snapshot-manager"
            status snapshot "$name"
            if [[ $source == "local" ]]; then
                $rsm $debug -d $name -k $zfs_auto_snapshot_keep -l $zfs_auto_snapshot_label
            else
                $rsm $debug -d $name -k $zfs_auto_snapshot_keep -l $zfs_auto_snapshot_label -h $source -p $sshport
            fi
        fi
    fi

    # source snapshots (oldest first), queried directly before the replication of this dataset
    source_snaps=$(remote "zfs list -H -o name,guid -s createtxg -t snapshot $name 2>/dev/null")
    if [ $zfs_auto_snapshot_keep -gt 1 ]; then
        leftovers=$(cut -f1 <<< "$source_snaps" | label_leftovers)
        if [[ -n $leftovers ]]; then
            log "[WARN] $name - snapshots on the source contain the label $zfs_auto_snapshot_label, but do not match the name scheme of the snapshot engine $zfs_auto_snapshot_engine (e.g. left over after switching the engine). Check and delete them on the source if they are no longer needed:\n$(sed 's/^/\t/' <<< "$leftovers")"
        fi
    fi
    if cut -f1 <<< "$source_snaps" | grep -qE "$snapshot_regex"; then
        # transfer of an earlier run interrupted (receive -s): continue it, otherwise discard the partial state,
        # it would block every further receive (an interrupted initial replication is removed completely)
        token=$(zfs get -H -o value receive_resume_token $target/$name 2>/dev/null)
        if [[ -n $token ]] && [[ $token != "-" ]]; then
            status send "$name"
            resumed=0
            if [[ $transfer_resume == "1" ]]; then
                log "[INFO] $name - resuming the interrupted transfer to $target/$name"
                # an interrupted initial replication has no snapshot yet
                if [[ -z $(zfs list -H -o name -t snapshot $target/$name 2>/dev/null) ]]; then
                    recv_opts=($nomount $mp $cm -x $tag -x com.sun:auto-snapshot)
                else
                    recv_opts=(-x $tag -x com.sun:auto-snapshot -F)
                fi
                if remote "zfs send $debug -t $token" | throttle | zfs receive -s "${recv_opts[@]}" $debug $target/$name; then
                    resumed=1
                    if [[ $fstype == "filesystem" ]] && [[ $(zfs get -H -o value mounted $target/$name) == "yes" ]]; then
                        zfs unmount $target/$name
                    fi
                else
                    log "[WARN] $name - resuming the transfer failed (e.g. the snapshot was deleted on the source), discarding the partial state."
                fi
            else
                log "[WARN] $name - $target/$name contains the partial state of an interrupted transfer, discarding it (transfer_resume=0)."
            fi
            if [ $resumed -eq 0 ]; then
                zfs receive -A $target/$name || { rc=1; log "[ERROR] discarding the partial state of $target/$name failed."; }
            fi
        fi

        IFS=$' '
        # initial replication: create the path on the target, then send the oldest matching snapshot
        if ! zfs list -H $target/$name > /dev/null 2>&1 ; then
            debug "$target/$name does not exist"
            prefix=""
            for part in $(echo $target/$(echo $name | cut -d'/' -f1) | sed "s/\// /g"); do
                if [ $(zfs list $prefix$part > /dev/null 2>&1 ; echo $?) -gt 0 ]; then
                    debug "$prefix$part does not exist"
                    log "[INFO] Creating $prefix$part"
                    if unmounted_target "${prefix%/}"; then
                        zfs create $autosnap -p $prefix$part
                    else
                        zfs create -o canmount=noauto $autosnap -p $prefix$part
                    fi
                fi
                prefix="$prefix$part/"
            done

            debug "$name - Start initial replication"
            status send "$name"
            IFS=$'\n'
            snap=$(remote "zfs list -H -t snapshot -o name -S createtxg $name | grep -E \"$snapshot_regex\" | tail -1")
            hold_snaps "$snap"
            if [[ -n $snap ]]; then
                log "[INFO] Start initial replication: $snap => $target/$(echo $name | cut -d'/' -f1)"
                if remote "zfs send -w -p $debug $snap" | throttle | zfs receive $resumable $nomount $mp $cm -x $tag -x com.sun:auto-snapshot $debug -dF $target/$(echo $name | cut -d'/' -f1); then
                    # receive mounts the new filesystem despite canmount=noauto: unmount it, so it does not show
                    # up as mounted filesystem (e.g. in checkmk) until the next reboot
                    if [[ $fstype == "filesystem" ]] && [[ $(zfs get -H -o value mounted $target/$name) == "yes" ]]; then
                        debug "Unmounting $target/$name"
                        zfs unmount $target/$name
                    fi
                else
                    rc=1; log "[ERROR] initial replication to $target/$name failed."
                fi
            fi
            # the initial replication may take long, query the source snapshots again
            source_snaps=$(remote "zfs list -H -o name,guid -s createtxg -t snapshot $name 2>/dev/null")
        fi

        # refquota: received only with the initial replication (send -p), a refquota raised on the source later
        # makes the receive fail with "quota exceeded". Only raise or remove it on the target, never lower it.
        if [[ $fstype == "filesystem" ]] && [[ ${refquotas[$name]} =~ ^[0-9]+$ ]]; then
            target_refquota=$(zfs get -Hp -o value refquota $target/$name 2>/dev/null)
            if [[ $target_refquota =~ ^[0-9]+$ ]] && [ $target_refquota -gt 0 ]; then
                if [ ${refquotas[$name]} -eq 0 ]; then
                    log "[INFO] $name - removing refquota $(size $target_refquota) of $target/$name (no refquota on the source)"
                    zfs set refquota=none $target/$name || { rc=1; log "[ERROR] setting refquota of $target/$name failed."; }
                elif [ ${refquotas[$name]} -gt $target_refquota ]; then
                    log "[INFO] $name - raising refquota of $target/$name from $(size $target_refquota) to $(size ${refquotas[$name]}) (refquota of the source)"
                    zfs set refquota=${refquotas[$name]} $target/$name || { rc=1; log "[ERROR] setting refquota of $target/$name failed."; }
                fi
            fi
        fi

        # incremental replication: from the newest target snapshot (matched by guid) to all newer source snapshots
        debug "$name - Start incremental replication"
        status send "$name"
        guid=$(zfs list -H -o guid -s createtxg -t snapshot $target/$name | tail -1)
        if [[ $guid != "" ]]; then
            last=$(grep $guid <<< "$source_snaps" | tail -1 | cut -f1)
            # newest target snapshot deleted on the source (rotation): continue from the bookmark of this snapshot.
            # A rollback on the source destroys newer bookmarks (zfs rollback -r), then the bookmark does not exist.
            delta_snaps=
            if [[ $last == "" ]] && [[ $source_bookmark == "1" ]]; then
                bookmark_txg=$(remote "zfs list -H -o name,guid,createtxg -t bookmark $name 2>/dev/null" \
                    | awk -F'\t' -v b="$name#$bookmark_name" -v g="$guid" '$1 == b && $2 == g { print $3 }')
                if [[ -n $bookmark_txg ]]; then
                    last="$name#$bookmark_name"
                    delta_snaps=$(remote "zfs list -H -o name,createtxg -s createtxg -t snapshot $name" \
                        | awk -F'\t' -v t="$bookmark_txg" '$2 > t { print $1 }' | grep -E "$snapshot_regex")
                    log "[INFO] $name - the newest snapshot of $target/$name was deleted on the source, continuing from the bookmark $last"
                fi
            fi
            # newest target snapshot missing on the source (rollback or deleted there): new chain from the newest
            # common snapshot, the old chain stays as branch (branch_replica)
            if [[ $last == "" ]]; then
                common=$(awk -F'\t' 'NR == FNR { g[$2] = 1; next } ($1 in g) { print $2; exit }' \
                    <(printf '%s\n' "$source_snaps") <(zfs list -H -o guid,name -S createtxg -t snapshot $target/$name))
                if [[ -n $common ]] && branch_replica "$name" "$common"; then
                    guid=$(zfs list -H -o guid -s createtxg -t snapshot $target/$name | tail -1)
                    last=$(grep $guid <<< "$source_snaps" | tail -1 | cut -f1)
                fi
            fi
            if [[ $last != "" ]]; then
                IFS=$'\n'
                if [[ $fstype == "filesystem" ]] && [[ $(zfs get -H -o value canmount $target/$name) != "noauto" ]] \
                    && ! unmounted_target $target/$name; then
                    zfs set canmount=noauto $target/$name
                fi
                if [[ -z $delta_snaps ]]; then
                    delta_snaps=$(grep -E "$snapshot_regex" <<< "$source_snaps" | grep --after-context=200 $guid | grep -v $guid | cut -f1)
                fi
                # base and all snapshots still to be sent stay on the source until this dataset is done
                hold_snaps "$last" $delta_snaps
                for snap in $delta_snaps; do
                    log "[INFO] Replicating delta of $last => $snap to $target/$name"
                    if ! remote "zfs send -w $debug -i $last $snap" | throttle | zfs receive $resumable -x $tag -x com.sun:auto-snapshot -F $debug $target/$name; then
                        # deleted on the source after the snapshots were listed (rotation, no hold): skip it, the next
                        # delta starts at the last received snapshot (an incremental stream can span deleted snapshots)
                        if ! remote "zfs list -H -o name -t snapshot $snap" > /dev/null 2>&1; then
                            log "[WARN] $name - $snap was deleted on the source during the replication, continuing with the next snapshot."
                            continue
                        fi
                        rc=1; log "[ERROR] incremental replication to $target/$name failed."
                    fi
                    last=$snap
                done
                # bookmark of the newest replicated snapshot for the next run (only replaced if it points elsewhere)
                if [[ $source_bookmark == "1" ]] && [ $rc -eq 0 ] && [[ $last == *@* ]]; then
                    debug "$name - bookmark $name#$bookmark_name => $last"
                    remote "if [ \"\$(zfs get -H -o value guid $name#$bookmark_name 2>/dev/null)\" != \"\$(zfs get -H -o value guid $last)\" ]; then zfs destroy $name#$bookmark_name 2>/dev/null; zfs bookmark $last $name#$bookmark_name; fi" \
                        || log "[WARN] $name - creating the bookmark $name#$bookmark_name failed, the replication continues without it."
                fi
                # replication failed: the chain breaks if the base is the oldest snapshot of its interval on the source
                if [ $rc -ne 0 ]; then
                    base=$(zfs list -H -o guid,name -s createtxg -t snapshot $target/$name | tail -1)
                    for interval in $(sed "s/|/\n/g" <<< "$snapshot_filter"); do
                        if grep -qE "${interval_regex[$interval]}" <<< "${base#*$'\t'}" \
                            && [[ $(grep -E "${interval_regex[$interval]}" <<< "$source_snaps" | head -1 | cut -f2) == "${base%%$'\t'*}" ]]; then
                            ds_state[$name]=at_risk
                            ds_detail[$name]="base ${base#*@} is the oldest snapshot of interval $interval on the source"
                            log "[ERROR] $name - the replication failed and its base ${base#*$'\t'} is the oldest snapshot of interval $interval on the source: the next rotation breaks the chain."
                        fi
                    done
                fi
            else
                log "[ERROR] No matching snapshot (guid: $guid) found on source filesystem. This can be outdated snapshots on target or a previously deleted and new created dataset $name on the source filesystem."
                if [ $rc -eq 0 ]; then rc=1; fi
                # newest common snapshot: target snapshots (newest first) whose guid exists on the source
                common=$(awk -F'\t' 'NR == FNR { g[$2] = 1; next } ($1 in g) { print $2; exit }' \
                    <(printf '%s\n' "$source_snaps") <(zfs list -H -o guid,name -S createtxg -t snapshot $target/$name))
                if [[ -n $common ]]; then
                    ds_state[$name]=diverged
                    ds_detail[$name]="newest common snapshot ${common#*@}"
                    log "[ERROR] $name - the newest snapshot on $target/$name does not exist on the source anymore (rollback or deleted on the source). Newest common snapshot: $common"
                else
                    # cause: dataset newer than the replica, or all snapshots of the replica rotated out on the source
                    target_newest=$(zfs list -Hp -o creation -S createtxg -t snapshot $target/$name | head -1)
                    source_times=$(remote "zfs get -Hp -o value creation $name; zfs list -Hp -o creation -s createtxg -t snapshot $name | head -1")
                    source_created=$(head -1 <<< "$source_times")
                    source_oldest=$(tail -1 <<< "$source_times")
                    if [ ${source_created:-0} -gt ${target_newest:-0} ]; then
                        cause="the dataset was created again on the source"
                    elif [ ${source_oldest:-0} -gt ${target_newest:-0} ]; then
                        cause="all snapshots of the replica were deleted by the rotation on the source"
                    else
                        cause="unknown cause"
                    fi
                    ds_state[$name]=broken
                    ds_detail[$name]=$cause
                    log "[ERROR] $name - no common snapshot with the source: $cause. A full replication is needed, to keep the old replica rename it and the next run replicates again:\n\t$(rename_hint $name)"
                fi
            fi
        else
            log "[ERROR] No snapshot found on $target/$name to add incremental snapshots to. The target dataset (with all children) needs to be deleted and recreated via replication."
            if [ $rc -eq 0 ]; then rc=1; fi
            ds_state[$name]=empty
            ds_detail[$name]="no snapshot on $target/$name"
            log "[ERROR] $name - to keep the empty replica rename it, the next run replicates again:\n\t$(rename_hint $name)"
        fi

        release_snaps

        # delete target snapshots: per label the newest target_keep[label] stay, otherwise those older than the oldest
        # source snapshot of the label are deleted (at least min_keep stay)
        debug "$name - Start deletion of old snapshots"
        status delete "$name"
        # source snapshots (newest first), queried again after the replication
        source_snaps=$(remote "zfs list -H -o guid,name -S createtxg -t snapshot $name")
        filter=$(echo -e $snapshot_filter | sed "s/|/\n/g")
        IFS=$'\n'
        for interval in $filter ; do
            debug "$name - Checking interval $interval"
            # fixed number on this host: the oldest beyond the newest target_keep snapshots, whatever the source deleted
            if [[ -n ${target_keep[$interval]} ]]; then
                target_snaps=$(zfs list -H -o name -S createtxg -t snapshot $target/$name | grep -E "${interval_regex[$interval]}")
                for snap in $(tail -n +$((target_keep[$interval] + 1)) <<< "$target_snaps"); do
                    clones=$(zfs get -H -o value clones $snap 2>/dev/null)
                    if [[ -n $clones && $clones != "-" ]]; then
                        log "[INFO] $name - keeping $snap, the old chain $clones depends on it."
                    else
                        log "[INFO] Deleting $snap (more than ${target_keep[$interval]} snapshots of $interval)"
                        zfs destroy $debug $snap
                    fi
                done
                continue
            fi
            guid=$(grep -E "${interval_regex[$interval]}" <<< "$source_snaps" | cut -f1 | tail -1)
            if [[ -n $guid ]]; then
                target_snaps=$(zfs list -H -o name,guid -S createtxg -t snapshot $target/$name | grep -E "${interval_regex[$interval]}")
                # deletion starts at the oldest source snapshot on the target: without it nothing is deleted
                if [[ -n $target_snaps ]] && ! grep -q $guid <<< "$target_snaps"; then
                    oldest=$(grep -E "${interval_regex[$interval]}" <<< "$source_snaps" | tail -1 | cut -f2)
                    # the replication itself is done: warning and dataset state for the monitoring, no exit code
                    log "[WARN] $name - the oldest source snapshot of interval $interval ($oldest) does not exist on $target/$name, no snapshots of this interval are deleted on the target."
                    if [[ -z ${ds_state[$name]} ]] || [[ ${ds_state[$name]} == "branched" ]]; then
                        ds_state[$name]=cleanup_blocked
                        ds_detail[$name]="oldest source snapshot ${oldest#*@} missing on the target"
                    elif [[ ${ds_state[$name]} == "cleanup_blocked" ]]; then
                        ds_detail[$name]+=", ${oldest#*@}"
                    fi
                fi
                snaps_to_delete=$(grep --after-context=200 $guid <<< "$target_snaps" | grep -v $guid | cut -f1)
                snap_count=$(grep -c . <<< "$target_snaps")

                for snap in $snaps_to_delete; do
                    clones=$(zfs get -H -o value clones $snap 2>/dev/null)
                    if [[ -n $clones && $clones != "-" ]]; then
                        # the old chain of a rollback on the source starts here (bashclub-zsync-maintenance)
                        log "[INFO] $name - keeping $snap, the old chain $clones depends on it."
                    elif [[ $snap_count -gt $min_keep ]]; then
                        log "[INFO] Deleting $snap"
                        debug "$name - snap_count=$snap_count, min_keep=$min_keep"
                        zfs destroy $debug $snap
                        ((snap_count--))
                    else
                        debug "$name - Skipping deletion of $snap. snap_count=$snap_count, min_keep=$min_keep"
                    fi
                done
            fi
        done
        if [ $zfs_auto_snapshot_keep -gt 1 ]; then
            leftovers=$(zfs list -H -o name -t snapshot $target/$name 2>/dev/null | label_leftovers)
            if [[ -n $leftovers ]]; then
                ds_leftovers[$name]=$(grep -c . <<< "$leftovers")
                log "[WARN] $name - snapshots on the target contain the label $zfs_auto_snapshot_label, but do not match the name scheme of the snapshot engine $zfs_auto_snapshot_engine. They are not deleted by zsync in strict mode, check and delete them if they are no longer needed:\n$(sed 's/^/\t/' <<< "$leftovers")"
            fi
        fi
    else
        debug "$name - No snapshots found with filter $snapshot_filter"
        ds_state[$name]=no_snapshots
    fi
    if [[ -z ${ds_state[$name]} ]] || { [[ ${ds_state[$name]} =~ ^(branched|cleanup_blocked)$ ]] && [ $rc -ne 0 ]; }; then
        if [ $rc -ne 0 ]; then ds_state[$name]=failed; else ds_state[$name]=ok; fi
    fi
    if [ $rc -ne 0 ]; then ((datasets_failed++)); fi
    if [ $rc_before -ne 0 ]; then rc=$rc_before; fi
    save_source_props "$name"
    log "[INFO] Replication of $name finished."
done

#### datasets only on the target: deleted or renamed on the source, or no longer selected for replication
# replicas are target datasets with snapshots (path datasets created by zsync have none); only checked if the
# datasets of the source are known, a failed query must not turn every replica into an orphan
# acknowledged with: zfs set bashclub:zsync-orphan=keep <replica>
if [ ${#fstypes[@]} -gt 0 ]; then
    status orphans
    declare -A selected=() kept=() used=() branches=()
    for n in "${syncvols[@]}"; do selected[$n]=1; done
    while IFS=$'\t' read -r n v; do
        if [[ -n $n ]] && [[ $v == "keep" ]]; then kept[$n]=1; fi
    done <<< "$(zfs get -H -o name,value -r -t filesystem,volume bashclub:zsync-orphan $target 2>/dev/null)"
    while IFS=$'\t' read -r n v; do
        if [[ -n $n ]]; then used[$n]=$v; fi
    done <<< "$(zfs list -H -o name,used -r $target 2>/dev/null)"
    # old chains after a rollback on the source (branch_replica), marked with bashclub:zsync-branch
    while IFS=$'\t' read -r n v; do
        if [[ -n $n ]]; then branches[$n]=$v; fi
    done <<< "$(zfs get -H -o name,value -s local -r -t filesystem,volume bashclub:zsync-branch $target 2>/dev/null)"
    for replica in $(zfs list -H -o name -t snapshot -r $target 2>/dev/null | cut -d'@' -f1 | uniq); do
        src=${replica#$target/}
        if [[ $replica == "$target" ]] || [[ -n ${selected[$src]} ]]; then continue; fi
        if [[ -n ${branches[$replica]} ]]; then
            ds_detail[$src]="old chain since ${branches[$replica]%%;*} (${branches[$replica]#*;}, ${used[$replica]})"
            if [[ -n ${kept[$replica]} ]]; then
                ds_state[$src]=branch_kept
            else
                ds_state[$src]=branch
                log "[WARN] $replica is the old chain after a rollback on the source: ${ds_detail[$src]}. Keep or delete it: bashclub-zsync-maintenance"
            fi
            continue
        fi
        if [[ -z ${fstypes[$src]} ]]; then
            ds_state[$src]=orphan_missing
            ds_detail[$src]="does not exist on the source"
        else
            ds_state[$src]=orphan_unselected
            ds_detail[$src]="is not selected for replication anymore ($tag)"
        fi
        if [[ -n ${kept[$replica]} ]]; then
            ds_state[$src]=orphan_kept
            debug "$replica ${ds_detail[$src]}, kept (bashclub:zsync-orphan=keep)"
        else
            log "[WARN] $replica ${ds_detail[$src]} and is not replicated anymore (${used[$replica]}). Keep it: zfs set bashclub:zsync-orphan=keep $replica - check the deletion: zfs destroy -rnv $replica"
        fi
    done
fi

# states per dataset, "since" is kept as long as the state of a dataset does not change
if [[ -n $status_file ]]; then
    declare -A since=()
    if [[ -f $status_lib.datasets ]]; then
        while IFS=$'\t' read -r n st t _; do since["$n|$st"]=$t; done < "$status_lib.datasets"
    fi
    for n in "${!ds_state[@]}"; do
        printf '%s\t%s\t%s\t%s\n' "$n" "${ds_state[$n]}" "${since["$n|${ds_state[$n]}"]:-$run_start}" "${ds_detail[$n]}"
    done | sort > "$status_lib.datasets.tmp" && mv -f "$status_lib.datasets.tmp" "$status_lib.datasets"
fi

#### open decisions for bashclub-zsync-maintenance as user property of the replica, so it reads all of them with one
# query: bashclub:zsync-case=<type>;<since>;<config>;<detail>; written only when a case arises or changes its type,
# removed when the case is resolved; datasets this run did not evaluate keep their marking
declare -A case_now=()
while IFS=$'\t' read -r n v; do
    if [[ -n $n ]]; then case_now[$n]=$v; fi
done <<< "$(zfs get -H -o name,value -s local -r -t filesystem,volume bashclub:zsync-case $target 2>/dev/null)"
for n in "${!ds_state[@]}"; do
    replica=$target/$n
    case ${ds_state[$n]} in
        orphan_missing|orphan_unselected) case_type=orphan ;;
        broken|empty) case_type=broken ;;
        branch) case_type=branch ;;
        *) if [[ -n ${ds_leftovers[$n]} ]]; then case_type=leftovers; else case_type=; fi ;;
    esac
    old=${case_now[$replica]}
    if [[ -z $case_type ]]; then
        if [[ -n $old ]]; then zfs inherit bashclub:zsync-case "$replica" 2>/dev/null; fi
        continue
    fi
    if [[ ${old%%;*} == "$case_type" ]]; then continue; fi
    if [[ $case_type == leftovers ]]; then detail="${ds_leftovers[$n]} snapshots of the label in another name scheme"
    else detail=${ds_detail[$n]}; fi
    zfs set "bashclub:zsync-case=$case_type;$(date +%s);$conf_path;${detail//;/,}" "$replica" 2>/dev/null \
        || log "[WARN] marking the open decision on $replica failed (bashclub:zsync-case), bashclub-zsync-maintenance does not show it."
done

#### monitoring: checkzfs at the end of the run, its output goes with the state into the spool file (on_exit)
if [ $checkzfs_disabled -eq 0 ]; then
    log "[INFO] Running checkzfs..."
    status checkzfs
    run_checkzfs
fi

exit $rc