diff --git a/Cargo.lock b/Cargo.lock index 16ed59b..c6b9e59 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1711,6 +1711,7 @@ dependencies = [ "fuser", "http", "libc", + "log", "notify", "prost", "sea-orm", diff --git a/Cargo.toml b/Cargo.toml index 2b5fc2c..3846970 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -45,6 +45,8 @@ http = "1" chrono = { version = "0.4", default-features = false, features = ["clock"] } tracing = "0.1" +# For sea_orm's ConnectOptions::sqlx_logging_level, which takes a log::LevelFilter. +log = "0.4" tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } # Pinned to 0.2.5: Builder::max_log_files (used for retention) landed in the # 0.2.3 Builder API and is current here. An unqualified "0.2" could otherwise diff --git a/devenv.nix b/devenv.nix index b2e360c..b6e89ab 100644 --- a/devenv.nix +++ b/devenv.nix @@ -29,6 +29,8 @@ just flac ffmpeg + id3v2 + mpv protobuf buf grpcurl @@ -39,7 +41,7 @@ processes.musicfs = { after = [ "devenv:processes:postgres" ]; - exec = '' + exec = lib.mkDefault '' cargo run -- \ --source /home/fujin/Music \ --mountpoint /tmp/rust-fuse \ @@ -47,6 +49,43 @@ ''; }; + profiles.local.module = { + processes.musicfs = { + after = [ "devenv:processes:postgres" ]; + exec = '' + cargo run -- \ + --source /home/fujin/Music \ + --mountpoint /tmp/rust-fuse \ + --database "postgresql://fujin@localhost/musicfs?host=$PGHOST" + ''; + }; + }; + + profiles.remote.module = { + processes.musicfs = { + after = [ "devenv:processes:postgres" ]; + exec = '' + cargo run -- \ + --source http://10.185.226.145:50051 \ + --mountpoint /tmp/rust-fuse \ + --database "postgresql://fujin@localhost/musicfs?host=$PGHOST" + ''; + }; + }; + + profiles.e2e.module = { + processes.musicfs = { + after = [ "devenv:processes:postgres" ]; + exec = '' + cargo run -- \ + --source http://127.0.0.1:50061 \ + --mountpoint ./target/e2e/mnt \ + --database "postgresql://fujin@localhost/musicfs_e2e?host=$PGHOST" \ + --log-dir ./logs/e2e + ''; + }; + }; + services.postgres = { enable = true; initialDatabases = [ @@ -54,6 +93,10 @@ name = "musicfs"; schema = ./db/schema.sql; } + { + name = "musicfs_e2e"; + schema = ./db/schema.sql; + } ]; }; diff --git a/justfile b/justfile index 027445c..7794c89 100644 --- a/justfile +++ b/justfile @@ -2,3 +2,9 @@ alias b := build build: cargo build + +up profile: + devenv up --profile {{profile}} + +e2e: + scripts/e2e/run.sh diff --git a/scripts/e2e/README.md b/scripts/e2e/README.md new file mode 100644 index 0000000..b66fb6b --- /dev/null +++ b/scripts/e2e/README.md @@ -0,0 +1,51 @@ +# musicfs E2E Tests + +Full distributed-path tests: server in Incus VM, local FUSE client, tests run +against the mountpoint. + +## Prerequisites + +- Inside `devenv shell` (provides ffmpeg, metaflac, id3v2, incus, grpcurl, Postgres) +- Postgres running (`devenv up postgres`) +- Incus installed and configured + +## Run + +```sh +just e2e # full suite (creates VM, runs tests, destroys VM) +KEEP=1 just e2e # keep VM + DB alive after run for debugging +``` + +## What it tests + +| Phase | Description | +| ------------ | ------------------------------------------------------- | +| structural | Virtual directory tree matches fixture tags | +| cache | First read = cache miss, second read = cache hit | +| FLAC read | metaflac reads tags through FUSE reassembly | +| FLAC write | metaflac writes vorbis comment, write handler persists | +| MP3 read | id3v2 reads ID3v2 tags through FUSE reassembly | +| MP3 write | id3v2 writes tags, write handler persists | +| chunked write| Non-metadata writes don't corrupt tag structure | +| persistence | Tags survive client restart (DB restore path) | + +## Fixtures + +`make-fixtures.sh` generates a deterministic library under `target/e2e/music/`: + +``` +E2E Artist A/E2E Album X/01 - Alpha.flac (with cover art) +E2E Artist A/E2E Album X/02 - Beta.flac +E2E Artist B/E2E Album Y/01 - Gamma.mp3 (with cover art) +E2E Artist B/E2E Album Y/02 - Delta.mp3 +``` + +All tags are known at test time → exact assertions. + +## Profiles + +```sh +just up e2e # start client against e2e VM (manual testing) +``` + +Client uses `musicfs_e2e` database and `./logs/e2e` for logs. diff --git a/scripts/e2e/lib.sh b/scripts/e2e/lib.sh new file mode 100644 index 0000000..f18db18 --- /dev/null +++ b/scripts/e2e/lib.sh @@ -0,0 +1,104 @@ +#!/usr/bin/env bash +# +# lib.sh — shared helpers for E2E test suite. +# +# Test functions set a global $FAILURES counter and $FAILED list so the +# caller can exit non-zero after all suites have run. +# +# Sourced by run.sh — do NOT execute directly. +# + +FAILURES=0 +PASSES=0 +FAILED_TESTS="" + +assert_eq() { + local desc="$1" expected="$2" actual="$3" + if [[ "$expected" == "$actual" ]]; then + printf ' \033[1;32mPASS\033[0m %s\n' "$desc" + ((++PASSES)) + else + printf ' \033[1;31mFAIL\033[0m %s\n' "$desc" + printf ' expected: %q\n' "$expected" + printf ' actual: %q\n' "$actual" + ((++FAILURES)) + FAILED_TESTS+="$desc; " + fi +} + +assert_contains() { + local desc="$1" haystack="$2" needle="$3" + if [[ "$haystack" == *"$needle"* ]]; then + printf ' \033[1;32mPASS\033[0m %s\n' "$desc" + ((++PASSES)) + else + printf ' \033[1;31mFAIL\033[0m %s\n' "$desc" + printf ' haystack did not contain: %q\n' "$needle" + ((++FAILURES)) + FAILED_TESTS+="$desc; " + fi +} + +assert_not_contains() { + local desc="$1" haystack="$2" needle="$3" + if [[ "$haystack" != *"$needle"* ]]; then + printf ' \033[1;32mPASS\033[0m %s\n' "$desc" + ((++PASSES)) + else + printf ' \033[1;31mFAIL\033[0m %s\n' "$desc" + printf ' haystack unexpectedly contained: %q\n' "$needle" + ((++FAILURES)) + FAILED_TESTS+="$desc; " + fi +} + +assert_match() { + local desc="$1" haystack="$2" regex="$3" + if [[ "$haystack" =~ $regex ]]; then + printf ' \033[1;32mPASS\033[0m %s\n' "$desc" + ((++PASSES)) + else + printf ' \033[1;31mFAIL\033[0m %s\n' "$desc" + printf ' did not match regex: %s\n' "$regex" + ((++FAILURES)) + FAILED_TESTS+="$desc; " + fi +} + +assert_file_exists() { + local desc="$1" path="$2" + if [[ -e "$path" ]]; then + printf ' \033[1;32mPASS\033[0m %s\n' "$desc" + ((++PASSES)) + else + printf ' \033[1;31mFAIL\033[0m %s\n' "$desc" + printf ' file not found: %s\n' "$path" + ((++FAILURES)) + FAILED_TESTS+="$desc; " + fi +} + +assert_gt() { + local desc="$1" a="$2" b="$3" + if (( a > b )); then + printf ' \033[1;32mPASS\033[0m %s (%s > %s)\n' "$desc" "$a" "$b" + ((++PASSES)) + else + printf ' \033[1;31mFAIL\033[0m %s (%s <= %s)\n' "$desc" "$a" "$b" + ((++FAILURES)) + FAILED_TESTS+="$desc; " + fi +} + +summary() { + echo + printf -- '---\n' + printf 'passes: %d\n' "$PASSES" + printf 'failures: %d\n' "$FAILURES" + if [[ -n "$FAILED_TESTS" ]]; then + printf 'failed: %s\n' "$FAILED_TESTS" + fi + if (( FAILURES > 0 )); then + return 1 + fi +} diff --git a/scripts/e2e/make-fixtures.sh b/scripts/e2e/make-fixtures.sh new file mode 100755 index 0000000..df639e9 --- /dev/null +++ b/scripts/e2e/make-fixtures.sh @@ -0,0 +1,88 @@ +#!/usr/bin/env bash +# +# make-fixtures.sh — generate a deterministic music library for E2E tests. +# +# Creates 2 FLAC + 2 MP3 files with known tags and embedded cover art under +# target/e2e/music/. The tags drive the FUSE virtual directory layout, so +# the directory names below are also the expected mount-point paths. +# +set -euo pipefail + +MUSIC_DIR="${1:-$(pwd)/target/e2e/music}" + +log() { printf '\033[1;34m[fixtures]\033[0m %s\n' "$*" >&2; } +die() { printf '\033[1;31m[fixtures error]\033[0m %s\n' "$*" >&2; exit 1; } + +command -v ffmpeg >/dev/null || die "ffmpeg not found (run inside devenv shell)" +command -v metaflac >/dev/null || die "metaflac not found (run inside devenv shell)" + +rm -rf "$MUSIC_DIR" +mkdir -p "$MUSIC_DIR" + +# ── cover art ────────────────────────────────────────────────────────── +COVER="$MUSIC_DIR/.cover.png" +COVER2="$MUSIC_DIR/.cover2.png" +ffmpeg -hide_banner -loglevel error -y \ + -f lavfi -i "color=c=0x884422:s=200x200:d=0.04" -frames:v 1 "$COVER" +ffmpeg -hide_banner -loglevel error -y \ + -f lavfi -i "color=c=0x224488:s=200x200:d=0.04" -frames:v 1 "$COVER2" + +# ── FLAC: E2E Artist A / E2E Album X ────────────────────────────────── +FLAC_DIR="$MUSIC_DIR/E2E Artist A/E2E Album X" +mkdir -p "$FLAC_DIR" + +log "generating FLAC Alpha …" +ffmpeg -hide_banner -loglevel error -y \ + -f lavfi -i "sine=frequency=440:duration=3" -c:a flac \ + -metadata title="Alpha" \ + -metadata artist="E2E Artist A" \ + -metadata album_artist="E2E Artist A" \ + -metadata album="E2E Album X" \ + -metadata tracknumber=1 \ + -metadata date=2024 \ + "$FLAC_DIR/01 - Alpha.flac" +metaflac --import-picture-from="3||||$COVER" "$FLAC_DIR/01 - Alpha.flac" + +log "generating FLAC Beta …" +ffmpeg -hide_banner -loglevel error -y \ + -f lavfi -i "sine=frequency=330:duration=3" -c:a flac \ + -metadata title="Beta" \ + -metadata artist="E2E Artist A" \ + -metadata album_artist="E2E Artist A" \ + -metadata album="E2E Album X" \ + -metadata tracknumber=2 \ + -metadata date=2024 \ + "$FLAC_DIR/02 - Beta.flac" + +# ── MP3: E2E Artist B / E2E Album Y ─────────────────────────────────── +MP3_DIR="$MUSIC_DIR/E2E Artist B/E2E Album Y" +mkdir -p "$MP3_DIR" + +log "generating MP3 Gamma …" +ffmpeg -hide_banner -loglevel error -y \ + -f lavfi -i "sine=frequency=392:duration=3" -i "$COVER2" \ + -c:a libmp3lame -q:a 4 -id3v2_version 3 \ + -metadata title="Gamma" \ + -metadata artist="E2E Artist B" \ + -metadata album_artist="E2E Artist B" \ + -metadata album="E2E Album Y" \ + -metadata tracknumber=1 \ + -metadata date=2023 \ + -map 0:a -map 1:v -disposition:v attached_pic \ + "$MP3_DIR/01 - Gamma.mp3" + +log "generating MP3 Delta …" +ffmpeg -hide_banner -loglevel error -y \ + -f lavfi -i "sine=frequency=294:duration=3" \ + -c:a libmp3lame -q:a 4 -id3v2_version 3 \ + -metadata title="Delta" \ + -metadata artist="E2E Artist B" \ + -metadata album_artist="E2E Artist B" \ + -metadata album="E2E Album Y" \ + -metadata tracknumber=2 \ + -metadata date=2023 \ + "$MP3_DIR/02 - Delta.mp3" + +rm -f "$COVER" "$COVER2" + +log "done — 4 files under $MUSIC_DIR" diff --git a/scripts/e2e/run.sh b/scripts/e2e/run.sh new file mode 100755 index 0000000..f3ebdc7 --- /dev/null +++ b/scripts/e2e/run.sh @@ -0,0 +1,358 @@ +#!/usr/bin/env bash +# +# run.sh — E2E test suite for musicfs. +# +# Full distributed path: server in Incus VM serving a generated music library +# over gRPC, local client mounting it as FUSE, tests run against the mountpoint. +# +# Requires: devenv shell (for ffmpeg, metaflac, id3v2, incus, grpcurl, Postgres) +# +# Usage: +# scripts/e2e/run.sh full suite (creates VM, runs tests, destroys) +# KEEP=1 scripts/e2e/run.sh keep VM + DB alive after run for debugging +# +set -euo pipefail + +# ─────────────────────────── config ──────────────────────────────────── +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +SCRIPTS_DIR="$(cd "$SCRIPT_DIR/.." && pwd)" +PROJECT_DIR="$(cd "$SCRIPT_DIR/../.." && pwd)" + +MNT="$PROJECT_DIR/target/e2e/mnt" +MUSIC_DIR="$PROJECT_DIR/target/e2e/music" +LOG_DIR="$PROJECT_DIR/logs/e2e" +DB_NAME="musicfs_e2e" +VM_NAME="musicfs-e2e" +PORT=50061 +KEEP="${KEEP:-0}" +RUST_LOG_VAL="${RUST_LOG:-musicfs=debug}" + +CLIENT_PID="" +STARTED_PG="" +. "$SCRIPT_DIR/lib.sh" + +log() { printf '\033[1;34m[e2e]\033[0m %s\n' "$*" >&2; } +die() { printf '\033[1;31m[e2e error]\033[0m %s\n' "$*" >&2; exit 1; } + +# ──────────────────────── postgres ───────────────────────────────────── +ensure_postgres() { + if pg_isready -h "$PGHOST" >/dev/null 2>&1; then return 0; fi + log "Postgres not running — starting via devenv…" + devenv up postgres & + STARTED_PG=$! + local i + for ((i = 0; i < 45; i++)); do + if pg_isready -h "$PGHOST" >/dev/null 2>&1; then + log "Postgres ready" + return 0 + fi + sleep 1 + done + die "Postgres did not become ready in 45s" +} + +stop_postgres() { + [[ -n "$STARTED_PG" ]] || return 0 + log "stopping Postgres (started by this script)…" + kill "$STARTED_PG" 2>/dev/null || true + wait "$STARTED_PG" 2>/dev/null || true + STARTED_PG="" +} + +# ──────────────────────── VM management ──────────────────────────────── +vm_ip() { incus list "$VM_NAME" -f csv -c 4 2>/dev/null | head -1 | awk '{print $1}'; } + +start_vm() { + log "starting VM '$VM_NAME' (port $PORT)…" + VM_NAME="$VM_NAME" \ + MUSIC_SOURCE="$MUSIC_DIR" \ + LISTEN_PORT="$PORT" \ + RUST_LOG=debug \ + bash "$SCRIPTS_DIR/vm.sh" up + + local ip; ip="$(vm_ip)" + [[ -n "$ip" ]] || die "could not determine VM IP" + log "VM IP: $ip" + + log "waiting for health check…" + local i + for ((i = 0; i < 45; i++)); do + if grpcurl -plaintext "$ip:$PORT" grpc.health.v1.Health/Check 2>/dev/null | grep -q SERVING; then + log "server healthy" + VM_IP="$ip" + return + fi + sleep 1 + done + die "server health check timed out" +} + +destroy_vm() { + VM_NAME="$VM_NAME" bash "$SCRIPTS_DIR/vm.sh" destroy 2>/dev/null || true +} + +# ──────────────────────── DB management ──────────────────────────────── +reset_db() { + log "resetting database '$DB_NAME'…" + dropdb --if-exists "$DB_NAME" 2>/dev/null || true + createdb "$DB_NAME" + psql -q -d "$DB_NAME" -f "$PROJECT_DIR/db/schema.sql" +} + +drop_db() { + dropdb --if-exists "$DB_NAME" 2>/dev/null || true +} + +# ────────────────────── client lifecycle ─────────────────────────────── +build_client() { + log "building client…" + (cd "$PROJECT_DIR" && cargo build --bin musicfs) +} + +start_client() { + mkdir -p "$MNT" "$LOG_DIR" + local dsn="postgresql://fujin@localhost/$DB_NAME?host=$PGHOST" + + log "starting client (mountpoint $MNT)…" + RUST_LOG="$RUST_LOG_VAL" \ + "$PROJECT_DIR/target/debug/musicfs" \ + --source "http://$VM_IP:$PORT" \ + --mountpoint "$MNT" \ + --database "$dsn" \ + --log-dir "$LOG_DIR" & + CLIENT_PID=$! + wait_for_mount +} + +wait_for_mount() { + log "waiting for FUSE mount…" + local i + for ((i = 0; i < 60; i++)); do + if [[ -d "$MNT/E2E Artist A" ]]; then + log "mount ready" + return + fi + if ! kill -0 "$CLIENT_PID" 2>/dev/null; then + die "client process exited prematurely — check $LOG_DIR" + fi + sleep 1 + done + die "FUSE mount not ready after 60s" +} + +stop_client() { + [[ -n "$CLIENT_PID" ]] || return 0 + kill "$CLIENT_PID" 2>/dev/null || true + local i + for ((i = 0; i < 10; i++)); do + kill -0 "$CLIENT_PID" 2>/dev/null || break + sleep 1 + done + kill -9 "$CLIENT_PID" 2>/dev/null || true + wait "$CLIENT_PID" 2>/dev/null || true + CLIENT_PID="" + fusermount -u "$MNT" 2>/dev/null || true + sleep 1 +} + +# ─────────────────────────── cleanup ─────────────────────────────────── +cleanup() { + local rv=$? + log "cleaning up…" + stop_client || true + if [[ "$KEEP" != "1" ]]; then + drop_db + destroy_vm + stop_postgres || true + rm -rf "$MNT" + else + log "KEEP=1 — VM, DB, and mount left intact for debugging" + log " mountpoint: $MNT" + log " VM IP: ${VM_IP:-}" + log " logs: $LOG_DIR" + fi + exit $rv +} +trap cleanup EXIT + +# ─────────────────────────── paths ───────────────────────────────────── +ALPHA="$MNT/E2E Artist A/E2E Album X/01 - Alpha.flac" +BETA="$MNT/E2E Artist A/E2E Album X/02 - Beta.flac" +GAMMA="$MNT/E2E Artist B/E2E Album Y/01 - Gamma.mp3" +DELTA="$MNT/E2E Artist B/E2E Album Y/02 - Delta.mp3" + +# ════════════════════════════════════════════════════════════════════════ +# TESTS +# ════════════════════════════════════════════════════════════════════════ + +test_structural() { + printf '\n\033[1;36m── structural ──\033[0m\n' + + local root; root="$(ls "$MNT" 2>/dev/null || true)" + assert_contains "root has Artist A dir" "$root" "E2E Artist A" + assert_contains "root has Artist B dir" "$root" "E2E Artist B" + + local album_a; album_a="$(ls "$MNT/E2E Artist A/E2E Album X" 2>/dev/null || true)" + assert_contains "Album X has Alpha.flac" "$album_a" "01 - Alpha.flac" + assert_contains "Album X has Beta.flac" "$album_a" "02 - Beta.flac" + + local album_b; album_b="$(ls "$MNT/E2E Artist B/E2E Album Y" 2>/dev/null || true)" + assert_contains "Album Y has Gamma.mp3" "$album_b" "01 - Gamma.mp3" + assert_contains "Album Y has Delta.mp3" "$album_b" "02 - Delta.mp3" + + local ftype; ftype="$(stat -c %F "$ALPHA" 2>/dev/null || echo "missing")" + assert_eq "Alpha is regular file" "regular file" "$ftype" + + local fsize; fsize="$(stat -c %s "$ALPHA" 2>/dev/null || echo 0)" + assert_gt "Alpha virtual size > 0" "$fsize" 0 + + local dtype; dtype="$(stat -c %F "$MNT/E2E Artist A" 2>/dev/null || echo "missing")" + assert_eq "Artist A is directory" "directory" "$dtype" +} + +test_cache() { + printf '\n\033[1;36m── cache behavior ──\033[0m\n' + + ffprobe -hide_banner -v quiet -show_format "$BETA" >/dev/null 2>&1 || true + ffprobe -hide_banner -v quiet -show_format "$BETA" >/dev/null 2>&1 || true + + sleep 1 + local log_file="" + log_file="$(ls -t "$LOG_DIR"/musicfs.*.log 2>/dev/null | head -1)" || true + local logs="" + if [[ -n "$log_file" ]]; then + logs="$(cat "$log_file" 2>/dev/null || true)" + fi + + assert_contains "cache miss on first read" "$logs" "cache miss" + assert_contains "cache hit on second read" "$logs" "cache hit" +} + +test_flac_read() { + printf '\n\033[1;36m── FLAC read ──\033[0m\n' + + local title; title="$(metaflac --show-tag=TITLE "$ALPHA" 2>/dev/null || true)" + local artist; artist="$(metaflac --show-tag=ARTIST "$ALPHA" 2>/dev/null || true)" + local album; album="$(metaflac --show-tag=ALBUM "$ALPHA" 2>/dev/null || true)" + + assert_contains "FLAC TITLE tag" "$title" "TITLE=Alpha" + assert_contains "FLAC ARTIST tag" "$artist" "ARTIST=E2E Artist A" + assert_contains "FLAC ALBUM tag" "$album" "ALBUM=E2E Album X" + + local pics; pics="$(metaflac --list --block-type=PICTURE "$ALPHA" 2>/dev/null || true)" + assert_contains "FLAC has cover art" "$pics" "PICTURE" +} + +test_flac_write() { + printf '\n\033[1;36m── FLAC write ──\033[0m\n' + + metaflac --remove-tag=ARTIST --set-tag="ARTIST=Modified FLAC" "$BETA" 2>/dev/null + + local artist; artist="$(metaflac --show-tag=ARTIST "$BETA" 2>/dev/null || true)" + assert_contains "FLAC write — ARTIST updated" "$artist" "ARTIST=Modified FLAC" + + local title; title="$(metaflac --show-tag=TITLE "$BETA" 2>/dev/null || true)" + assert_contains "FLAC write — TITLE preserved" "$title" "TITLE=Beta" +} + +test_mp3_read() { + printf '\n\033[1;36m── MP3 read ──\033[0m\n' + + local info; info="$(id3v2 --list "$GAMMA" 2>/dev/null || true)" + + assert_contains "MP3 TITLE tag" "$info" "Gamma" + assert_contains "MP3 ARTIST tag" "$info" "E2E Artist B" + assert_contains "MP3 ALBUM tag" "$info" "E2E Album Y" +} + +test_mp3_write() { + printf '\n\033[1;36m── MP3 write ──\033[0m\n' + + id3v2 --song "Modified Delta" --artist "Modified MP3" "$DELTA" 2>/dev/null + + local info; info="$(id3v2 --list "$DELTA" 2>/dev/null || true)" + assert_contains "MP3 write — TIT2 updated" "$info" "Modified Delta" + assert_contains "MP3 write — TPE1 updated" "$info" "Modified MP3" +} + +test_chunked_write() { + printf '\n\033[1;36m── chunked write resilience ──\033[0m\n' + + # Write at offset 8192 — well past any metadata blocks for both formats. + # The write handler should ignore it (no tag pattern match) and the + # metadata structure remains parseable. + + local title_before; title_before="$(metaflac --show-tag=TITLE "$ALPHA" 2>/dev/null || true)" + printf 'XXXX' | dd of="$ALPHA" bs=1 seek=8192 count=4 conv=notrunc 2>/dev/null || true + sleep 0.5 + local title_after; title_after="$(metaflac --show-tag=TITLE "$ALPHA" 2>/dev/null || true)" + assert_eq "FLAC tags unchanged after non-metadata write" "$title_before" "$title_after" + + local mp3_before; mp3_before="$(id3v2 --list "$GAMMA" 2>/dev/null | grep -i 'title' || true)" + printf 'YYYY' | dd of="$GAMMA" bs=1 seek=8192 count=4 conv=notrunc 2>/dev/null || true + sleep 0.5 + local mp3_after; mp3_after="$(id3v2 --list "$GAMMA" 2>/dev/null | grep -i 'title' || true)" + assert_eq "MP3 tags unchanged after non-metadata write" "$mp3_before" "$mp3_after" +} + +test_persistence() { + printf '\n\033[1;36m── persistence (client restart) ──\033[0m\n' + + metaflac --remove-tag=ARTIST --set-tag="ARTIST=Persisted Artist" "$ALPHA" 2>/dev/null + + local before; before="$(metaflac --show-tag=ARTIST "$ALPHA" 2>/dev/null || true)" + assert_contains "persistence — write before restart" "$before" "ARTIST=Persisted Artist" + + log "restarting client…" + stop_client + start_client + + local after; after="$(metaflac --show-tag=ARTIST "$ALPHA" 2>/dev/null || true)" + assert_contains "persistence — read after restart" "$after" "ARTIST=Persisted Artist" +} + +# ════════════════════════════════════════════════════════════════════════ +# MAIN +# ════════════════════════════════════════════════════════════════════════ + +main() { + [[ -n "${PGHOST:-}" ]] || die "PGHOST not set — run inside devenv shell" + ensure_postgres + command -v incus >/dev/null || die "incus not found" + command -v grpcurl >/dev/null || die "grpcurl not found" + command -v metaflac >/dev/null || die "metaflac not found" + command -v id3v2 >/dev/null || die "id3v2 not found" + command -v ffprobe >/dev/null || die "ffprobe not found" + + log "=== E2E suite start ===" + + # Phase 0: fixtures + bash "$SCRIPT_DIR/make-fixtures.sh" "$MUSIC_DIR" + + # Phase 1: server + start_vm + + # Phase 2: client + build_client + reset_db + start_client + + # Phase 3-8: tests (read-only first, mutating last) + test_structural + test_cache + test_flac_read + test_flac_write + test_mp3_read + test_mp3_write + test_chunked_write + test_persistence + + # Summary + printf '\n' + summary + local rv=$? + log "=== E2E suite done ===" + exit $rv +} + +main "$@" diff --git a/src/db/cache.rs b/src/db/cache.rs index 1650323..6ec338b 100644 --- a/src/db/cache.rs +++ b/src/db/cache.rs @@ -25,7 +25,7 @@ pub async fn put_cached_bytes(inode: i64, data: Vec, db: &DatabaseConnection let active = ActiveModel { inode: Set(inode), data: Set(data), - fetched_at: Set(chrono::Utc::now().naive_utc()), + fetched_at: Set(chrono::Utc::now()), }; if let Err(e) = Entity::insert(active) .on_conflict( @@ -55,3 +55,61 @@ pub async fn delete_cached_bytes_for(inodes: &[i64], db: &DatabaseConnection) { error!(error = %e, "delete_cached_bytes_for: DB delete failed"); } } + +#[cfg(test)] +mod tests { + use super::*; + use sea_orm::ConnectionTrait; + + async fn connect() -> Option { + let pg_host = std::env::var("PGHOST").ok()?; + let url = format!("postgresql://fujin@localhost/musicfs?host={pg_host}"); + let db = sea_orm::Database::connect(&url).await.ok()?; + db.execute_unprepared("SELECT 1 FROM cached_file_bytes LIMIT 0") + .await + .ok()?; + Some(db) + } + + async fn setup_parent_item(db: &DatabaseConnection, inode: i64) { + db.execute_unprepared(&format!( + "INSERT INTO items (inode, name, original_path, local_path, file_type, hash) \ + VALUES ({inode}, 'test', '/test', '/test', 'file', 0) \ + ON CONFLICT (inode) DO NOTHING" + )) + .await + .unwrap(); + } + + async fn teardown(db: &DatabaseConnection, inode: i64) { + let _ = db + .execute_unprepared(&format!("DELETE FROM items WHERE inode = {inode}")) + .await; + } + + #[tokio::test] + async fn cache_round_trip_with_timestamptz() { + let db = match connect().await { + Some(db) => db, + None => { + eprintln!("skip: no database (PGHOST unset or unreachable)"); + return; + } + }; + + let inode = -999_999_i64; + let test_data = b"regression-test-payload".to_vec(); + + setup_parent_item(&db, inode).await; + put_cached_bytes(inode, test_data.clone(), &db).await; + + let cached = get_cached_bytes(inode, &db).await; + assert_eq!( + cached, + Some(test_data), + "get_cached_bytes must decode fetched_at (TIMESTAMPTZ), not fail with type mismatch" + ); + + teardown(&db, inode).await; + } +} diff --git a/src/db/entities.rs b/src/db/entities.rs index c313144..67bcea7 100644 --- a/src/db/entities.rs +++ b/src/db/entities.rs @@ -46,7 +46,7 @@ pub mod cached_file_bytes { pub inode: i64, #[sea_orm(column_type = "Blob")] pub data: Vec, - pub fetched_at: DateTime, + pub fetched_at: DateTimeUtc, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/src/main.rs b/src/main.rs index ba047a2..c088742 100644 --- a/src/main.rs +++ b/src/main.rs @@ -56,7 +56,11 @@ async fn main() { "musicfs starting" ); - let db = sea_orm::Database::connect(&args.database) + // sea-orm logs every statement at INFO by default, which floods normal + // output; demote to DEBUG so queries stay hidden under RUST_LOG=info. + let mut db_opts = sea_orm::ConnectOptions::new(args.database.clone()); + db_opts.sqlx_logging_level(log::LevelFilter::Debug); + let db = sea_orm::Database::connect(db_opts) .await .unwrap_or_else(|e| { error!(database = %args.database, error = %e, "database connect failed"); @@ -71,7 +75,9 @@ async fn main() { error!(source = %args.source, error = %e, "network origin init failed"); panic!("network origin init: {e}"); }); - let snapshot = origin.snapshot().unwrap_or_else(|e| { + // Await directly on this runtime; the sync `snapshot()` wrapper would + // try to build a nested runtime and panic (we're inside #[tokio::main]). + let snapshot = origin.snapshot_async().await.unwrap_or_else(|e| { error!(source = %args.source, error = %e, "network initial snapshot failed"); panic!("network initial snapshot: {e}"); }); @@ -105,12 +111,13 @@ async fn main() { let files: Arc>> = Arc::new(std::sync::Mutex::new(snapshot)); - watcher.watch(files.clone()); + let watcher_handle = watcher.watch(files.clone()); let fs = FuseFs { files, bytes: byte_source, client: db, + runtime_handle: tokio::runtime::Handle::current(), }; let cfg = fuser::Config::default(); @@ -135,6 +142,7 @@ async fn main() { } info!(mountpoint = %mountpoint, "unmounting"); + watcher_handle.stop(); drop(session); } diff --git a/src/music/mp3.rs b/src/music/mp3.rs index 2d4ed30..8020358 100644 --- a/src/music/mp3.rs +++ b/src/music/mp3.rs @@ -253,7 +253,7 @@ fn parse_id3_tag_frames(data: &[u8], out: &mut MusicMetadata) { } } -/// MP3 encoder. Builds a fresh ID3v2.4 tag from the current (possibly +/// MP3 encoder. Builds a fresh ID3v2.3 tag from the current (possibly /// overridden) tag fields. Preserved frames are not copied into the header; /// their lengths are counted so the tag's size field spans them. pub struct Mp3MusicMetadataEncoder; @@ -274,21 +274,24 @@ fn syncsafe_encode(n: u64) -> [u8; 4] { ] } -/// Build a single UTF-8 ID3v2.4 text frame. +/// Build a single UTF-16 ID3v2.3 text frame. fn build_id3_text_frame(id: &[u8; 4], text: &str) -> Vec { - let mut body = Vec::with_capacity(text.len() + 1); - body.push(0x03); // text encoding: UTF-8 - body.extend_from_slice(text.as_bytes()); + let mut body = Vec::with_capacity(3 + text.len() * 2); + body.push(0x01); // text encoding: UTF-16 with BOM + body.extend_from_slice(&[0xFF, 0xFE]); // little-endian BOM + for unit in text.encode_utf16() { + body.extend_from_slice(&unit.to_le_bytes()); + } let mut frame = Vec::with_capacity(10 + body.len()); frame.extend_from_slice(id); - frame.extend_from_slice(&syncsafe_encode(body.len() as u64)); + frame.extend_from_slice(&(body.len() as u32).to_be_bytes()); frame.extend_from_slice(&[0, 0]); // frame flags frame.extend_from_slice(&body); frame } -/// Build the in-memory portion of the virtual ID3v2.4 tag: the 10-byte tag +/// Build the in-memory portion of the virtual ID3v2.3 tag: the 10-byte tag /// header plus our four override frames. Preserved frames and audio are /// stitched in by the read path; the tag size field accounts for them. fn build_id3v2_header(m: &MusicMetadata) -> Vec { @@ -309,7 +312,7 @@ fn build_id3v2_header(m: &MusicMetadata) -> Vec { let mut header = Vec::with_capacity(10 + frames.len()); header.extend_from_slice(b"ID3"); - header.extend_from_slice(&[0x04, 0x00, 0x00]); // v2.4.0, no flags + header.extend_from_slice(&[0x03, 0x00, 0x00]); // v2.3.0, no flags header.extend_from_slice(&syncsafe_encode(body_len)); header.extend_from_slice(&frames); header @@ -444,7 +447,7 @@ mod tests { let header = build_id3v2_header(&mm); assert_eq!(&header[0..3], b"ID3"); - assert_eq!(&header[3..6], &[0x04, 0x00, 0x00]); + assert_eq!(&header[3..6], &[0x03, 0x00, 0x00]); let frames_len = header.len() as u64 - 10; let declared = syncsafe([header[6], header[7], header[8], header[9]]); diff --git a/src/origins/local/file_io.rs b/src/origins/local/file_io.rs index 1d34e54..3d71694 100644 --- a/src/origins/local/file_io.rs +++ b/src/origins/local/file_io.rs @@ -14,15 +14,7 @@ pub fn read_bytes_at(path: &Path, offset: u64, len: usize) -> io::Result return Ok(buf); } -/// Reassemble a virtualized FLAC byte range by stitching together the cached -/// header, picture block headers, and on-demand audio/picture bytes fetched -/// through `reader`. -/// -/// `reader(offset, len)` is transport-agnostic: for the local origin it wraps -/// `read_bytes_at(&path, ...)`; a network origin will route the same call -/// through its byte cache + HTTP range request. The FLAC assembly logic never -/// touches a `Path` directly. -pub fn assemble_flac_read( +fn assemble_virtual_read( reader: &dyn Fn(u64, usize) -> io::Result>, header: &[u8], pic_hdrs: &[Vec], @@ -34,7 +26,6 @@ pub fn assemble_flac_read( let end = offset + size as u64; let header_end = header.len() as u64; - // Pure header read — most common for tag readers. if end <= header_end { return Ok(header[offset as usize..end as usize].to_vec()); } @@ -85,6 +76,46 @@ pub fn assemble_flac_read( return Ok(buf); } +pub fn assemble_flac_read( + reader: &dyn Fn(u64, usize) -> io::Result>, + header: &[u8], + pic_hdrs: &[Vec], + pic_ranges: &[(u64, u64)], + real_audio_start: u64, + offset: u64, + size: u32, +) -> io::Result> { + assemble_virtual_read( + reader, + header, + pic_hdrs, + pic_ranges, + real_audio_start, + offset, + size, + ) +} + +pub fn assemble_mp3_read( + reader: &dyn Fn(u64, usize) -> io::Result>, + header: &[u8], + pic_hdrs: &[Vec], + pic_ranges: &[(u64, u64)], + real_audio_start: u64, + offset: u64, + size: u32, +) -> io::Result> { + assemble_virtual_read( + reader, + header, + pic_hdrs, + pic_ranges, + real_audio_start, + offset, + size, + ) +} + #[cfg(test)] mod tests { use super::*; @@ -159,4 +190,26 @@ mod tests { let expected = [b'C', b'D', 100, 101, 102, 103]; assert_eq!(result, expected); } + + #[test] + fn assemble_mp3_read_header_only() { + let header = b"ID3\x04\x00DATAHERE"; + let noop = |_, _| Ok(vec![]); + let result = assemble_mp3_read(&noop, header, &[], &[], 0, 4, 4).unwrap(); + assert_eq!(result, &header[4..8]); + } + + #[test] + fn assemble_mp3_read_audio_region() { + let mut f = tempfile::NamedTempFile::new().unwrap(); + let content: Vec = (0..20).collect(); + f.write_all(&content).unwrap(); + f.flush().unwrap(); + let path = f.path().to_path_buf(); + + let reader = file_reader(path); + let header = b"ID3"; + let result = assemble_mp3_read(&reader, header, &[], &[], 10, 5, 4).unwrap(); + assert_eq!(result, vec![12, 13, 14, 15]); + } } diff --git a/src/origins/local/watcher.rs b/src/origins/local/watcher.rs index 67b4e1e..bd40821 100644 --- a/src/origins/local/watcher.rs +++ b/src/origins/local/watcher.rs @@ -9,7 +9,7 @@ use fuser::INodeNo; use notify::{Event, EventKind, RecursiveMode, Watcher}; use crate::item::Item; -use crate::origins::FileWatcher; +use crate::origins::{FileWatcher, WatcherHandle}; use tracing::{debug, error, info, trace}; pub struct LocalOriginFileWatcher { @@ -27,7 +27,7 @@ impl LocalOriginFileWatcher { } impl FileWatcher for LocalOriginFileWatcher { - fn watch(&self, files: Arc>>) { + fn watch(&self, files: Arc>>) -> WatcherHandle { let source = self.source.clone(); let destination = self.destination.clone(); @@ -75,5 +75,6 @@ impl FileWatcher for LocalOriginFileWatcher { } } }); + return WatcherHandle::detached(); } } diff --git a/src/origins/mod.rs b/src/origins/mod.rs index cc70fe0..d2645a8 100644 --- a/src/origins/mod.rs +++ b/src/origins/mod.rs @@ -14,7 +14,6 @@ use fuser::{Errno, FileAttr, Filesystem, Generation, INodeNo, Request}; use sea_orm::{ActiveModelTrait, DatabaseConnection}; use tracing::{debug, error, trace}; -use crate::db::sync::run_db_blocking; use crate::item::{FileType, Item}; use crate::music::db::save_music_metadata; use crate::origins::local::file_io; @@ -37,7 +36,35 @@ pub trait ByteSource: Send + Sync { } pub trait FileWatcher: Send + Sync { - fn watch(&self, files: Arc>>); + fn watch(&self, files: Arc>>) -> WatcherHandle; +} + +/// Returned by [`FileWatcher::watch`] so the caller can stop a background +/// watcher before the process (and its tokio runtime) shuts down. Without +/// this, an async watcher blocked on a timer/stream panics when the runtime is +/// torn down out from under it. +pub struct WatcherHandle { + stop: Option>, +} + +impl WatcherHandle { + pub fn new(stop: impl FnOnce() + Send + 'static) -> Self { + return WatcherHandle { + stop: Some(Box::new(stop)), + }; + } + + /// A watcher with no shutdown work — its thread exits on its own. + pub fn detached() -> Self { + return WatcherHandle { stop: None }; + } + + /// Signal the watcher to stop and wait for it to finish. + pub fn stop(mut self) { + if let Some(stop) = self.stop.take() { + stop(); + } + } } pub trait Origin: Send + Sync { @@ -50,6 +77,7 @@ pub struct FuseFs { pub files: Arc>>, pub bytes: Arc, pub client: DatabaseConnection, + pub runtime_handle: tokio::runtime::Handle, } pub(crate) fn file_to_attr(item: &Item) -> FileAttr { @@ -192,7 +220,7 @@ impl Filesystem for FuseFs { debug!(%ino, "write: ID3v2 tag update detected"); Some(mm.clone()) } - Some(mm) if !mm.header.is_empty() && data.len() == 128 && &data[0..3] == b"TAG" => { + Some(mm) if mm.header.is_empty() && data.len() == 128 && &data[0..3] == b"TAG" => { mm.update_from_id3v1_data(data); debug!(%ino, "write: ID3v1 tag update detected"); Some(mm.clone()) @@ -204,7 +232,7 @@ impl Filesystem for FuseFs { if let Some(music_metadata) = updated_music_metadata { let client = self.client.clone(); let ino_i64 = ino.0 as i64; - run_db_blocking(async move { + self.runtime_handle.block_on(async move { if let Err(e) = save_music_metadata(ino_i64, &music_metadata, &client).await { error!(ino = ino_i64, error = %e, "write: save_music_metadata failed"); } else { @@ -373,7 +401,7 @@ impl Filesystem for FuseFs { drop(files); - run_db_blocking(async move { + self.runtime_handle.block_on(async move { use sea_orm::ActiveValue::Set; if let Err(e) = (crate::db::entities::ActiveModel { inode: Set(inode_i64), @@ -403,7 +431,7 @@ impl Filesystem for FuseFs { reply: fuser::ReplyData, ) { trace!(%ino, offset, size, "read"); - let (inode, locator, flac) = { + let (inode, locator, virtual_layout) = { let files = self.files.lock().unwrap(); let item = match files.get(&ino) { Some(f) => f, @@ -415,22 +443,23 @@ impl Filesystem for FuseFs { }; let inode = item.inode; let locator = item.original_path.clone(); - let flac = item + let virtual_layout = item .music_metadata .as_ref() .filter(|mm| !mm.header.is_empty()) .map(|mm| { ( + mm.header.starts_with(b"ID3"), mm.header.clone(), mm.picture_block_headers.clone(), mm.picture_data_ranges.clone(), mm.real_audio_start, ) }); - (inode, locator, flac) + (inode, locator, virtual_layout) }; - let Some((header, pic_hdrs, pic_ranges, real_audio_start)) = flac else { + let Some((is_mp3, header, pic_hdrs, pic_ranges, real_audio_start)) = virtual_layout else { match self.bytes.read_at(inode, &locator, offset, size as usize) { Ok(bytes) => reply.data(&bytes), Err(e) => { @@ -443,18 +472,31 @@ impl Filesystem for FuseFs { let bytes = &self.bytes; let reader = |off: u64, len: usize| bytes.read_at(inode, &locator, off, len); - match file_io::assemble_flac_read( - &reader, - &header, - &pic_hdrs, - &pic_ranges, - real_audio_start, - offset, - size, - ) { + let result = if is_mp3 { + file_io::assemble_mp3_read( + &reader, + &header, + &pic_hdrs, + &pic_ranges, + real_audio_start, + offset, + size, + ) + } else { + file_io::assemble_flac_read( + &reader, + &header, + &pic_hdrs, + &pic_ranges, + real_audio_start, + offset, + size, + ) + }; + match result { Ok(bytes) => reply.data(&bytes), Err(e) => { - error!(%inode, offset, size, error = %e, "read: flac assembly failed; returning EIO"); + error!(%inode, offset, size, error = %e, "read: assembly failed; returning EIO"); reply.error(Errno::EIO); } } diff --git a/src/origins/network/mod.rs b/src/origins/network/mod.rs index f88fda1..0266813 100644 --- a/src/origins/network/mod.rs +++ b/src/origins/network/mod.rs @@ -29,7 +29,11 @@ use self::transport::NetworkTransport; pub struct NetworkOrigin { pub(crate) endpoint: String, pub(crate) destination: PathBuf, - runtime: tokio::runtime::Runtime, + /// Handle to the process-wide runtime (main's `#[tokio::main]`). Used by the + /// sync FUSE byte/watcher paths to drive async transport via `block_on`. + /// A Handle (not an owned Runtime) so dropping NetworkOrigin in async + /// context can't panic, and the runtime outlives any clones we hand out. + handle: tokio::runtime::Handle, transport: NetworkTransport, client: sea_orm::DatabaseConnection, /// In-memory cache of the latest manifest, keyed by server id. Written by @@ -54,15 +58,13 @@ impl NetworkOrigin { destination: String, client: sea_orm::DatabaseConnection, ) -> io::Result { - let runtime = tokio::runtime::Builder::new_multi_thread() - .enable_all() - .build() - .map_err(io_err)?; + // Must be called from within the process runtime (main's #[tokio::main]). + let handle = tokio::runtime::Handle::current(); let transport = NetworkTransport::new(endpoint.clone()).map_err(io_err)?; return Ok(NetworkOrigin { endpoint, destination: destination.into(), - runtime, + handle, transport, client, latest_manifest: Arc::new(RwLock::new(BTreeMap::new())), @@ -70,7 +72,7 @@ impl NetworkOrigin { } pub fn runtime_handle(&self) -> tokio::runtime::Handle { - return self.runtime.handle().clone(); + return self.handle.clone(); } } @@ -99,7 +101,11 @@ impl Origin for NetworkOrigin { } impl NetworkOrigin { - async fn snapshot_async(&self) -> io::Result> { + /// Async snapshot driven directly on the caller's runtime. `main` (already + /// `#[tokio::main]`) awaits this; the sync `Origin::snapshot()` wrapper is + /// only for non-async callers and must not be invoked from within a runtime + /// (it builds and `block_on`s a throwaway one). + pub async fn snapshot_async(&self) -> io::Result> { // 1. Pull client's current (inode, hash) pairs from the DB. let client_entries: Vec<(u64, u64)> = item_entities::Entity::find() .all(&self.client) @@ -129,21 +135,30 @@ impl NetworkOrigin { changed_or_deleted.extend(response.deleted.iter().map(|i| *i as i64)); delete_cached_bytes_for(&changed_or_deleted, &self.client).await; - // 4. Fetch fresh metadata for changed inodes only. - let wanted: Vec = response.changed.iter().map(|ih| ih.inode).collect(); + // latest_manifest is in-memory; on restart it's empty so the reconcile + // delta misses unchanged files. Fetch full manifest then, delta otherwise. let mut current_manifest = self.latest_manifest.read().unwrap().clone(); - if !wanted.is_empty() { - let wanted_count = wanted.len(); - let entries = self.transport.get_metadata(wanted).await.map_err(|e| { - error!( - wanted = wanted_count, - error = %e, - "network snapshot: get_metadata from server failed" - ); + if current_manifest.is_empty() { + let entries = self.transport.get_manifest().await.map_err(|e| { + error!(error = %e, "network snapshot: get_manifest from server failed"); io_err(e) })?; - for entry in entries { - current_manifest.insert(entry.id, entry); + current_manifest = entries.into_iter().map(|e| (e.id, e)).collect(); + } else { + let wanted: Vec = response.changed.iter().map(|ih| ih.inode).collect(); + if !wanted.is_empty() { + let wanted_count = wanted.len(); + let entries = self.transport.get_metadata(wanted).await.map_err(|e| { + error!( + wanted = wanted_count, + error = %e, + "network snapshot: get_metadata from server failed" + ); + io_err(e) + })?; + for entry in entries { + current_manifest.insert(entry.id, entry); + } } } for inode in &response.deleted { @@ -317,7 +332,9 @@ impl ByteSource for NetworkByteSource { let inode_i64 = inode.0 as i64; trace!(%inode, offset, len, "network read_at"); - let cached = run_db_blocking(get_cached_bytes(inode_i64, &self.client)); + let cached = self + .runtime_handle + .block_on(get_cached_bytes(inode_i64, &self.client)); if let Some(data) = cached { debug!(%inode, "read_at cache hit"); return slice_range(&data, offset, len); @@ -331,7 +348,8 @@ impl ByteSource for NetworkByteSource { error!(%inode, error = %e, "read_at: fetch_file_range failed"); io_err(e) })?; - run_db_blocking(put_cached_bytes(inode_i64, data.clone(), &self.client)); + self.runtime_handle + .block_on(put_cached_bytes(inode_i64, data.clone(), &self.client)); return slice_range(&data, offset, len); } diff --git a/src/origins/network/transport.rs b/src/origins/network/transport.rs index 7aa4f8d..cbd2e38 100644 --- a/src/origins/network/transport.rs +++ b/src/origins/network/transport.rs @@ -5,8 +5,8 @@ use tokio_stream::StreamExt; use tonic::codec::Streaming; use crate::proto::{ - ChangeEvent, GetFileRequest, GetMetadataRequest, InodeHash, ManifestEntry, MusicFsClient, - ReconcileRequest, ReconcileResponse, SubscribeEventsRequest, + ChangeEvent, GetFileRequest, GetManifestRequest, GetMetadataRequest, InodeHash, ManifestEntry, + MusicFsClient, ReconcileRequest, ReconcileResponse, SubscribeEventsRequest, }; /// Thin wrapper over the generated tonic client. Owns the connection and @@ -63,6 +63,20 @@ impl NetworkTransport { return Ok(out); } + pub async fn get_manifest(&self) -> Result> { + let mut client = self.client.lock().await; + let mut stream: Streaming = client + .get_manifest(GetManifestRequest {}) + .await + .context("GetManifest RPC failed")? + .into_inner(); + let mut out = Vec::new(); + while let Some(entry) = stream.next().await { + out.push(entry.context("GetManifest stream error")?); + } + return Ok(out); + } + pub async fn fetch_file_range( &self, id: u64, diff --git a/src/origins/network/watcher.rs b/src/origins/network/watcher.rs index b9f0d39..8bf1d7e 100644 --- a/src/origins/network/watcher.rs +++ b/src/origins/network/watcher.rs @@ -8,11 +8,12 @@ use std::{ use fuser::INodeNo; use sea_orm::{DatabaseConnection, EntityTrait}; +use tokio::sync::Notify; use tokio_stream::StreamExt; use crate::item::Item; -use crate::origins::FileWatcher; use crate::origins::network::transport::NetworkTransport; +use crate::origins::{FileWatcher, WatcherHandle}; use crate::proto::ManifestEntry as ProtoManifestEntry; use tracing::{info, warn}; @@ -51,8 +52,9 @@ impl NetworkOriginFileWatcher { /// The trait signature still requires it for parity with LocalOriginFileWatcher; /// a future refactor can rebuild the map in place here. impl FileWatcher for NetworkOriginFileWatcher { - fn watch(&self, files: Arc>>) { + fn watch(&self, files: Arc>>) -> WatcherHandle { let runtime_handle = self.runtime_handle.clone(); + let shutdown = Arc::new(Notify::new()); let state = WatcherState { transport: self.transport.clone(), client: self.client.clone(), @@ -60,8 +62,13 @@ impl FileWatcher for NetworkOriginFileWatcher { latest_manifest: self.latest_manifest.clone(), files, }; - thread::spawn(move || { - runtime_handle.block_on(state.run_loop()); + let loop_shutdown = shutdown.clone(); + let join = thread::spawn(move || { + runtime_handle.block_on(state.run_loop(loop_shutdown)); + }); + return WatcherHandle::new(move || { + shutdown.notify_one(); + let _ = join.join(); }); } } @@ -76,22 +83,33 @@ struct WatcherState { } impl WatcherState { - async fn run_loop(self) { + async fn run_loop(self, shutdown: Arc) { loop { - match self.transport.subscribe_events().await { + let subscribed = tokio::select! { + biased; + _ = shutdown.notified() => return, + s = self.transport.subscribe_events() => s, + }; + match subscribed { Ok(mut stream) => { info!("network watcher: subscribed to /events"); - while let Some(item) = stream.next().await { + loop { + let item = tokio::select! { + biased; + _ = shutdown.notified() => return, + item = stream.next() => item, + }; match item { - Ok(_event) => { + Some(Ok(_event)) => { if let Err(e) = self.reconcile_once().await { warn!(error = %e, "network watcher: reconcile after event failed"); } } - Err(e) => { + Some(Err(e)) => { warn!(error = %e, "network watcher: stream error; reconnecting"); break; } + None => break, } } } @@ -99,7 +117,11 @@ impl WatcherState { warn!(error = %e, "network watcher: subscribe failed; will retry"); } } - tokio::time::sleep(POLL_FALLBACK_INTERVAL).await; + tokio::select! { + biased; + _ = shutdown.notified() => return, + _ = tokio::time::sleep(POLL_FALLBACK_INTERVAL) => {} + } if let Err(e) = self.reconcile_once().await { warn!(error = %e, "network watcher: poll reconcile failed"); } diff --git a/src/server/transport/grpc.rs b/src/server/transport/grpc.rs index 3420c1c..86838a4 100644 --- a/src/server/transport/grpc.rs +++ b/src/server/transport/grpc.rs @@ -204,11 +204,22 @@ impl MusicFsTrait for MusicFsService { Status::internal(format!("read file range: {e}")) })?; - let chunk_stream = tokio_stream::iter(vec![Ok::(FileChunk { - data: chunk, - offset: start, - total_size, - })]); + // Split into <=1MB messages: gRPC's default max is 4MB and music files + // routinely exceed it. + const CHUNK_SIZE: usize = 1024 * 1024; + let chunk_stream = tokio_stream::iter( + chunk + .chunks(CHUNK_SIZE) + .enumerate() + .map(|(i, c)| { + Ok::(FileChunk { + data: c.to_vec(), + offset: start + (i * CHUNK_SIZE) as u64, + total_size, + }) + }) + .collect::>(), + ); return Ok(Response::new(Box::pin(chunk_stream))); }