Migrate an Ipfs Repo Off a Failing Disk

#!/usr/bin/env bash
# Prints how the old and the new repo on <host>, then this workstation's own repo, shard their block files;
# the copy needs all three alike.
# usage: check_sharding.sh <host> <old repo> <new repo>
set -o nounset

ssh -n "$1" cat "$2/blocks/SHARDING" "$3/blocks/SHARDING"
cat "${IPFS_PATH:-${HOME}/.ipfs}/blocks/SHARDING"

#!/usr/bin/env bash
# Makes a new kubo repo at <repo> on <host>, owned by the ssh user.
# usage: init_repo.sh <host> <repo>
set -o nounset

ssh -n "$1" "docker run --rm --user \"\$(id -u):\$(id -g)\" -v $2:/data/ipfs --entrypoint ipfs \
    ipfs/kubo:v0.39.0 init --profile server"

#!/usr/bin/env bash
# Cuts the cids read on stdin that no batch on <host> holds yet into batches of 1000, numbered after the last.
# usage: cut_batches.sh <host> <migration dir> < cids, bare or as /ipfs/<cid>
set -o nounset

host="$1"
dir="$2"
next="$(ssh -n "${host}" "mkdir -p ${dir}/batches && ls ${dir}/batches | wc -l")"
new="$(comm -13 <(ssh -n "${host}" "cat ${dir}/batches/* 2> /dev/null" | sort -u) \
    <(grep -v '^$' | sed 's|^\([^/]\)|/ipfs/\1|' | sort -u))"
if test -n "${new}"
then
    echo "${new}" | ssh "${host}" split --lines=1000 --numeric-suffixes="${next}" --suffix-length=3 - \
        "${dir}/batches/"
fi
echo "$(echo -n "${new}" | grep -c .) new cids, from batch ${next}"

#!/usr/bin/env python3
"""List the flatfs block files of whole DAGs, reading the block files of a repo directly, without ipfs.

usage: list_blocks.py <blocks dir> < lines of /ipfs/<cid>
prints one <shard>/<KEY>.data per block, each once; complains on stderr about blocks it can't find or read.

A dag-pb block holds its links (field 2) before its data (field 1), so its first byte tells whether it has
children: 0x12 means links follow, anything else means a leaf. Leaves, which hold nearly all the bytes,
are thus never read past their first byte. raw blocks have no children and are not opened at all.
"""
import base64
import sys

DAG_PB = 0x70
RAW = 0x55
IDENTITY = 0x00
B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"


def varint(data, pos):
    value = shift = 0
    while True:
        byte = data[pos]
        pos += 1
        value |= (byte & 0x7F) << shift
        if not byte & 0x80:
            return value, pos
        shift += 7


def b58decode(text):
    number = 0
    for char in text:
        number = number * 58 + B58.index(char)
    raw = number.to_bytes((number.bit_length() + 7) // 8, "big")
    return b"\0" * (len(text) - len(text.lstrip("1"))) + raw


def parse_binary_cid(data):
    """(codec, multihash) of a binary cid"""
    if len(data) == 34 and data[0] == 0x12 and data[1] == 0x20:
        return DAG_PB, data
    version, pos = varint(data, 0)
    if version != 1:
        raise ValueError(f"cid version {version}")
    codec, pos = varint(data, pos)
    return codec, data[pos:]


def parse_text_cid(text):
    if text.startswith("Qm"):
        return DAG_PB, b58decode(text)
    if text.startswith("b"):
        padded = text[1:].upper() + "=" * (-len(text[1:]) % 8)
        return parse_binary_cid(base64.b32decode(padded))
    raise ValueError(f"unsupported cid {text}")


def block_file(multihash):
    # flatfs next-to-last/2: the key is the base32 multihash, sharded by its next-to-last two characters
    key = base64.b32encode(multihash).decode().rstrip("=")
    return f"{key[-3:-1]}/{key}.data"


def links(content):
    """binary cids of a dag-pb node's links"""
    pos = 0
    while pos < len(content):
        tag, pos = varint(content, pos)
        if tag & 7 != 2:
            raise ValueError(f"dag-pb field {tag >> 3} of wire type {tag & 7}")
        length, pos = varint(content, pos)
        field = content[pos:pos + length]
        pos += length
        if tag >> 3 != 2:
            continue
        inner = 0
        while inner < len(field):
            inner_tag, inner = varint(field, inner)
            if inner_tag & 7 == 2:
                inner_length, inner = varint(field, inner)
                if inner_tag >> 3 == 1:
                    yield field[inner:inner + inner_length]
                inner += inner_length
            elif inner_tag & 7 == 0:
                _, inner = varint(field, inner)
            else:
                raise ValueError(f"dag-pb link wire type {inner_tag & 7}")


def main():
    blocks = sys.argv[1]
    seen = set()
    failed = False
    todo = [parse_text_cid(line.strip().removeprefix("/ipfs/")) for line in sys.stdin if line.strip()]
    while todo:
        codec, multihash = todo.pop()
        if multihash[0] == IDENTITY:
            # inlined in its cid, no block file
            continue
        path = block_file(multihash)
        if path in seen:
            continue
        seen.add(path)
        print(path)
        if codec == RAW:
            continue
        if codec != DAG_PB:
            print(f"unsupported codec {codec:#x}: {path}", file=sys.stderr)
            failed = True
            continue
        try:
            with open(f"{blocks}/{path}", "rb") as block:
                if block.read(1) != b"\x12":
                    continue
                block.seek(0)
                content = block.read()
            todo.extend(parse_binary_cid(child) for child in links(content))
        except (OSError, ValueError, IndexError) as error:
            print(f"cannot read {path}: {error}", file=sys.stderr)
            failed = True
    sys.exit(1 if failed else 0)


main()

#!/usr/bin/env bash
# Times reading one batch's block files on <host> in hash order, then in their order on disk.
# usage: read_order.sh <host> <old blocks dir> <migration dir> <batch>
set -o nounset

ssh "$1" bash -s "$2" "$3" "$4" <<'EOF'
cd "$1" || exit 1
files="$(python3 "$2/list_blocks.py" . < "$2/batches/$3" 2> /dev/null)"
hash_order="$(echo "${files}" | sed -n '1,800p')"
# first physical block of each file, from its extent map
disk_order="$(echo "${files}" | sed -n '801,1600p' | xargs /usr/sbin/filefrag -e \
    | awk '/^File size of/ { file = $4 } /^ +0:/ { sub(/\.\./, "", $4); print $4, file }' \
    | sort -n | cut -d' ' -f2)"

paused="$(pgrep -f "^python3 $2/copy_sorted" | tr '\n' ' ')"
test -n "${paused}" && kill -STOP ${paused}
trap 'test -n "${paused}" && kill -CONT ${paused}' EXIT

for order in hash_order disk_order
do
    list="${!order}"
    bytes="$(echo "${list}" | tr '\n' '\0' | du -cb --files0-from=- | tail -n 1 | cut -f1)"
    start="$(date +%s%N)"
    echo "${list}" | xargs cat > /dev/null
    ms=$(( ($(date +%s%N) - start) / 1000000 ))
    echo "${order}: $(echo "${list}" | wc -l) files, $((bytes / 1000000)) MB in ${ms} ms," \
         "$(( bytes / 1000 / (ms + 1) )) MB/s"
done
EOF

#!/usr/bin/env python3
"""Copy block files from one flatfs to another, reading them in the order they lie on the source disk.

usage: copy_sorted.py <source blocks dir> <target blocks dir> <file of shard/KEY.data lines>

Read in hash order, the head seeks before each file, which halves what the failing disk reads; sorted by their
first physical block, reads mostly follow each other. Files already in the target are skipped. Each file is
written under a temporary name then renamed, so a block file is whole or absent. Prints how many files and
bytes were copied, and each file it could not read; exits non zero if any.
"""
import os
import subprocess
import sys
import time

source, target, listing = sys.argv[1:4]
with open(listing) as lines:
    wanted = [line.strip() for line in lines if line.strip()]
missing = [path for path in wanted if not os.path.exists(os.path.join(target, path))]


def physical_start(paths):
    """first physical block of each file, from filefrag's extent maps; unknown ones go last"""
    start = {}
    for index in range(0, len(paths), 500):
        chunk = paths[index:index + 500]
        result = subprocess.run(["/usr/sbin/filefrag", "-e", *chunk], cwd=source, capture_output=True,
                                text=True)
        current = None
        for line in result.stdout.splitlines():
            if line.startswith("File size of "):
                current = line[len("File size of "):].split(" is ")[0]
            elif current is not None and line.strip().startswith("0:"):
                start[current] = int(line.split(":")[2].split("..")[0])
                current = None
    return start


began = time.monotonic()
start = physical_start(missing)
missing.sort(key=lambda path: start.get(path, float("inf")))
print(f"sorted {len(missing)} missing files by place on disk in {time.monotonic() - began:.0f} s", flush=True)
began = time.monotonic()

copied = size = 0
failed = []
for path in missing:
    destination = os.path.join(target, path)
    try:
        with open(os.path.join(source, path), "rb") as block:
            content = block.read()
    except OSError as error:
        failed.append(f"{path}: {error}")
        continue
    os.makedirs(os.path.dirname(destination), exist_ok=True)
    with open(destination + ".tmp", "wb") as out:
        out.write(content)
    os.rename(destination + ".tmp", destination)
    copied += 1
    size += len(content)

for failure in failed:
    print(f"could not read {failure}")
print(f"copied {copied} files, {size} bytes in {time.monotonic() - began:.0f} s; {len(wanted) - len(missing)}"
      f" already there, {len(failed)} unreadable")
sys.exit(1 if failed else 0)

#!/usr/bin/env bash
# Samples the given disks of <host> twenty seconds apart: busy share, reads, mean read time, writes.
# usage: disk_load.sh <host> <disk>...
set -o nounset

host="$1"
shift
pattern=" ($(IFS='|'; echo "$*")) "
ssh -n "${host}" "grep -E '${pattern}' /proc/diskstats; sleep 20; grep -E '${pattern}' /proc/diskstats" \
    | awk '{ name = $3 }
           name in reads { printf "%s: busy %d%%, read %d MB in %d reads of %d ms each, wrote %d MB\n",
                             name, ($13 - busy[name]) / 200, ($6 - sread[name]) / 2048,
                             $4 - reads[name], ($7 - rms[name]) / ($4 - reads[name] + 1),
                             ($10 - swritten[name]) / 2048; next }
           { reads[name] = $4; sread[name] = $6; rms[name] = $7; swritten[name] = $10; busy[name] = $13 }'

#!/usr/bin/env python3
"""Check that each listed block file is in the target flatfs with the same size as in the source.

usage: check_present.py <source blocks dir> <target blocks dir> <file of shard/KEY.data lines>

A light check run right after a copy: it reads only file sizes, never block contents, so it costs the target
disk nothing beside the copy. Prints each missing or short file; exits non zero if any. Whether every DAG is
whole is left to dag stat --offline, run over all batches once the source no longer needs reading.
"""
import os
import sys

source, target, listing = sys.argv[1:4]
bad = 0
with open(listing) as lines:
    for path in (line.strip() for line in lines):
        if not path:
            continue
        try:
            copied = os.stat(os.path.join(target, path)).st_size
        except FileNotFoundError:
            print(f"missing: {path}")
            bad += 1
            continue
        try:
            original = os.stat(os.path.join(source, path)).st_size
        except OSError as error:
            print(f"cannot stat source {path}: {error}")
            bad += 1
            continue
        if copied != original:
            print(f"size {copied} instead of {original}: {path}")
            bad += 1
print(f"{bad} block files missing or short")
sys.exit(1 if bad else 0)

#!/usr/bin/env bash
# Copies every batch not yet copied, one after the other: list its block files, copy them in disk order, check
# each copied file is there with its size. Once the old repo needs no more reading, checks every batch not yet
# whole offline in the new repo, marking whole batches imported and listing the cids that are not in
# copy/broken. Run detached; it logs to copy/all.log.
# usage: copy_all.sh <migration dir> <old repo> <new repo>
set -o nounset
set -o pipefail

dir="$1"
old="$2"
new="$3"
kubo="ipfs/kubo:v0.39.0"

mkdir -p "${dir}/copy" "${dir}/copied" "${dir}/imported"
exec >> "${dir}/copy/all.log" 2>&1

log () {
    echo "$(date -Is) $*"
}

copy () {
    local name="$1" work="${dir}/copy/$1"
    mkdir -p "${work}"
    python3 "${dir}/list_blocks.py" "${old}/blocks" < "${dir}/batches/${name}" > "${work}/files" \
        || log "${name}: some blocks could not be listed, see above"
    log "${name}: $(wc -l < "${work}/files") block files"
    # unreadable files are listed and left for the check to name the cids they break
    python3 "${dir}/copy_sorted.py" "${old}/blocks" "${new}/blocks" "${work}/files" \
        || log "${name}: some block files could not be read"
    # sizes only: reading the copied bytes back here would compete with the copy for the new disk
    if python3 "${dir}/check_present.py" "${old}/blocks" "${new}/blocks" "${work}/files"
    then log "${name}: copied, every block file present"
    else log "${name}: copied, some block files missing, see above"
    fi
}

# marks the batch imported only when every one of its cids is whole in the new repo
check () {
    local name="$1" result
    result="$(docker run --rm -i --user "$(id -u):$(id -g)" -v "${new}:/data/ipfs" --entrypoint sh "${kubo}" -c '
        broken=0
        while read -r path
        do
            if ! error="$(ipfs dag stat --offline --progress=false "${path#/ipfs/}" 2>&1 > /dev/null)"
            then broken=$((broken + 1)); echo "not whole: ${path}: ${error}"
            fi
        done
        echo "broken ${broken}"
    ' < "${dir}/batches/${name}")"
    echo "${result}" | grep "^not whole"
    if test "$(echo "${result}" | tail -n 1)" = "broken 0"
    then
        echo whole-by-dag-stat > "${dir}/imported/${name}"
        log "${name}: checked, all whole"
    else
        echo "${result}" | sed -n "s|^not whole: \([^:]*\):.*|${name} \1|p" >> "${dir}/copy/broken"
        log "${name}: checked, $(echo "${result}" | tail -n 1 | cut -d' ' -f2) cids not whole, listed in broken"
    fi
}

for batch in "${dir}"/batches/*
do
    name="$(basename "${batch}")"
    test -e "${dir}/imported/${name}" -o -e "${dir}/copied/${name}" && continue
    copy "${name}"
    touch "${dir}/copied/${name}"
done
log "all batches copied, the old repo no longer read; checking every DAG in the new one"

# lists only what this check finds, so a cid rescued since drops out
: > "${dir}/copy/broken"
for batch in "${dir}"/batches/*
do
    name="$(basename "${batch}")"
    test -e "${dir}/imported/${name}" && continue
    check "${name}"
done
log "all batches checked; $(ls "${dir}/imported" | wc -l) of $(ls "${dir}/batches" | wc -l) whole," \
    "$(wc -l < "${dir}/copy/broken") cids to rescue, listed in copy/broken"

#!/usr/bin/env bash
# Puts the copy scripts in <migration dir> on <host>, then starts copy_all.sh there detached, unless it runs.
# usage: start_chain.sh <host> <migration dir> <old repo> <new repo>
set -o nounset

host="$1"
dir="$2"
here="$(dirname "$(readlink -f "$0")")"
ssh -n "${host}" mkdir -p "${dir}"
rsync -a "${here}/list_blocks.py" "${here}/copy_sorted.py" "${here}/check_present.py" "${here}/copy_all.sh" \
    "${host}:${dir}/"
if ssh -n "${host}" pgrep -f "'^bash ${dir}/copy_all.sh'" > /dev/null
then
    echo "copy_all.sh already runs"
else
    ssh -n "${host}" "setsid nohup bash ${dir}/copy_all.sh ${dir} $3 $4 > /dev/null 2>&1 < /dev/null &"
    echo "copy_all.sh started"
fi

#!/usr/bin/env bash
# Tells where the chain on <host> is: its last log lines, how many batches are whole, whether it runs.
# usage: chain_status.sh <host> <migration dir>
set -o nounset

dir="$2"
ssh -n "$1" "tail -n 5 ${dir}/copy/all.log; \
    echo whole: \$(ls ${dir}/imported | wc -l) of \$(ls ${dir}/batches | wc -l) batches; \
    pgrep -f '^bash ${dir}/copy_all.sh' > /dev/null && echo running || echo ended"

#!/usr/bin/env bash
# Gets each cid listed in copy/broken on <host> into this workstation's node, then copies its block files into
# the new repo there.
# usage: rescue_broken.sh <host> <migration dir> <new repo> [retrieve command, given each /ipfs/<cid>]
set -o nounset

host="$1"
dir="$2"
new="$3"
shift 3
here="$(dirname "$(readlink -f "$0")")"
blocks="${IPFS_PATH:-${HOME}/.ipfs}/blocks"
for path in $(ssh -n "${host}" "cut -d' ' -f2 ${dir}/copy/broken" | sort -u)
do
    if ! { test $# -gt 0 && "$@" "${path}" < /dev/null > /dev/null 2>&1; } \
            && ! ipfs dag stat --progress=false "${path#/ipfs/}" < /dev/null > /dev/null
    then
        echo "${path}: neither retrieved nor on the swarm"
        continue
    fi
    echo "${path}" | python3 "${here}/list_blocks.py" "${blocks}" \
        | rsync -a --ignore-existing --files-from=- "${blocks}/" "${host}:${new}/blocks/"
    echo "${path}: blocks copied from this workstation"
done