Some more fixes for reconcile with remote

justfile

@@ -6,5 +6,14 @@ build:
 up profile:
     devenv up --profile {{profile}}
 
+vm command:
+    scripts/vm.sh {{command}}
+    
+redeploy:
+    scripts/vm.sh redeploy
+
 e2e:
-    scripts/e2e/run.sh    
+    scripts/e2e/run.sh
+
+e2e-resilience:
+    scripts/e2e/run-resilience.sh    

scripts/e2e/README.md

@@ -14,10 +14,13 @@ against the mountpoint.
 ```sh
 just e2e              # full suite (creates VM, runs tests, destroys VM)
 KEEP=1 just e2e       # keep VM + DB alive after run for debugging
+just e2e-resilience   # resilience suite (server stop/start, network partition, outage recovery)
 ```
 
 ## What it tests
 
+### Functional suite (`just e2e`)
+
 | Phase        | Description                                              |
 | ------------ | ------------------------------------------------------- |
 | structural   | Virtual directory tree matches fixture tags             |
@@ -29,6 +32,15 @@ KEEP=1 just e2e       # keep VM + DB alive after run for debugging
 | chunked write| Non-metadata writes don't corrupt tag structure         |
 | persistence  | Tags survive client restart (DB restore path)            |
 
+### Resilience suite (`just e2e-resilience`)
+
+| Scenario          | Disruption                          | What it verifies                                         |
+| ----------------- | ----------------------------------- | -------------------------------------------------------- |
+| server stop/start | `systemctl stop/start` in VM        | EIO on uncached reads, cached reads survive, watcher resubscribes within 10s |
+| network partition | `incus pause/start` (silent drop)   | EIO within 10s (tonic 3s timeout × kernel readahead), watcher detects via HTTP/2 keepalive within 30s |
+| cache resilience  | server stop, all files pre-cached   | All reads/writes from cache, no server contact needed    |
+| long outage       | `incus pause` for 15s+              | Watcher detects via keepalive PING, resubscribes within 10s of recovery |
+
 ## Fixtures
 
 `make-fixtures.sh` generates a deterministic library under `target/e2e/music/`:

scripts/e2e/resilience-helpers.sh

@@ -0,0 +1,309 @@
+#!/usr/bin/env bash
+#
+# resilience-helpers.sh — disruption primitives and FUSE/log assertions for
+# network stability and reconnection E2E tests.
+#
+# Requires lib.sh globals ($FAILURES, $PASSES, $FAILED_TESTS) and config
+# variables from run-resilience.sh ($VM_NAME, $PORT, $VM_IP, $MNT, $LOG_DIR,
+# $DB_NAME, $PGHOST, $CLIENT_PID, $ALPHA, $BETA, $GAMMA, $DELTA).
+#
+# Sourced by run-resilience.sh — do NOT execute directly.
+
+# ── timeouts (seconds) ──
+HEALTH_TIMEOUT=45
+READ_TIMEOUT=15
+READ_FAIL_TIMEOUT=10
+RECOVER_TIMEOUT=30
+VM_RESUME_TIMEOUT=30
+
+# ── internal pass/fail (reuses lib.sh globals) ──
+_res_pass() {
+  printf '  \033[1;32mPASS\033[0m %s\n' "$1"
+  ((++PASSES))
+}
+
+_res_fail() {
+  printf '  \033[1;31mFAIL\033[0m %s\n' "$1"
+  ((++FAILURES))
+  FAILED_TESTS+="$1; "
+}
+
+# ── client liveness ──
+check_client_alive() {
+  if [[ -n "$CLIENT_PID" ]] && ! kill -0 "$CLIENT_PID" 2>/dev/null; then
+    printf '\033[1;31m[e2e error]\033[0m client process died — check %s\n' "$LOG_DIR" >&2
+    return 1
+  fi
+  return 0
+}
+
+# ── disruption primitives ──
+server_stop() {
+  log "stopping server in VM…"
+  incus exec "$VM_NAME" -- systemctl stop musicfs-server.service 2>/dev/null || true
+}
+
+server_start() {
+  log "starting server in VM…"
+  incus exec "$VM_NAME" -- systemctl start musicfs-server.service 2>/dev/null || true
+}
+
+vm_pause() {
+  log "pausing VM '$VM_NAME'…"
+  incus pause "$VM_NAME"
+}
+
+vm_resume() {
+  log "resuming VM '$VM_NAME'…"
+  incus start "$VM_NAME"
+  local i
+  for ((i = 0; i < VM_RESUME_TIMEOUT; i++)); do
+    if incus exec "$VM_NAME" -- true 2>/dev/null; then return 0; fi
+    sleep 1
+  done
+  die "VM did not become reachable after resume"
+}
+
+# ── health check helpers ──
+wait_health() {
+  local timeout="${1:-$HEALTH_TIMEOUT}"
+  local i
+  for ((i = 0; i < timeout; i++)); do
+    if timeout 5 grpcurl -plaintext "$VM_IP:$PORT" grpc.health.v1.Health/Check 2>/dev/null | grep -q SERVING; then
+      return 0
+    fi
+    sleep 1
+  done
+  return 1
+}
+
+assert_health_down() {
+  local desc="$1"
+  if timeout 5 grpcurl -plaintext "$VM_IP:$PORT" grpc.health.v1.Health/Check 2>/dev/null | grep -q SERVING; then
+    _res_fail "$desc (server still serving)"
+  else
+    _res_pass "$desc"
+  fi
+}
+
+assert_health_up() {
+  local desc="$1"
+  if wait_health "$HEALTH_TIMEOUT"; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc (health check timed out)"
+  fi
+}
+
+# ── cache control ──
+clear_cache() {
+  log "clearing Postgres byte cache…"
+  psql -q -d "$DB_NAME" -c "DELETE FROM cached_file_bytes;" 2>/dev/null || true
+}
+
+# ── prewarm ──
+prewarm_file() {
+  local file="$1"
+  timeout "$READ_TIMEOUT" dd if="$file" of=/dev/null bs=4096 2>/dev/null || true
+}
+
+prewarm_all() {
+  log "prewarming cache (reading all test files)…"
+  prewarm_file "$ALPHA"
+  prewarm_file "$BETA"
+  prewarm_file "$GAMMA"
+  prewarm_file "$DELTA"
+}
+
+# ── FUSE read assertions ──
+assert_fuse_read_ok() {
+  local desc="$1" file="$2"
+  check_client_alive || return 1
+  if timeout "$READ_TIMEOUT" dd if="$file" of=/dev/null bs=4096 2>/dev/null; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc (read failed or timed out after ${READ_TIMEOUT}s)"
+  fi
+}
+
+# Assert a FUSE read fails with EIO *before* max_seconds.
+# Distinguishes "fast EIO" (system returned an error) from "indefinite hang"
+# (timeout killed dd). Currently FAILS during silent partitions because
+# there's no tonic timeout configured — the RPC hangs until the OS gives up.
+assert_fuse_eio_within() {
+  local desc="$1" file="$2" max_seconds="${3:-$READ_FAIL_TIMEOUT}"
+  check_client_alive || return 1
+  local start; start=$(date +%s)
+  local rc=0
+  timeout "$max_seconds" dd if="$file" of=/dev/null bs=4096 2>/dev/null || rc=$?
+  local end; end=$(date +%s)
+  local elapsed=$((end - start))
+  if (( rc == 0 )); then
+    _res_fail "$desc (read succeeded unexpectedly in ${elapsed}s)"
+  elif (( elapsed >= max_seconds )); then
+    _res_fail "$desc (hung ${elapsed}s — no EIO returned; missing tonic timeout)"
+  else
+    _res_pass "$desc (EIO in ${elapsed}s)"
+  fi
+}
+
+assert_fuse_metaflac_ok() {
+  local desc="$1" file="$2"
+  check_client_alive || return 1
+  if timeout "$READ_TIMEOUT" metaflac --show-tag=TITLE "$file" >/dev/null 2>&1; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc (metaflac read failed)"
+  fi
+}
+
+assert_fuse_id3v2_ok() {
+  local desc="$1" file="$2"
+  check_client_alive || return 1
+  if timeout "$READ_TIMEOUT" id3v2 --list "$file" >/dev/null 2>&1; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc (id3v2 read failed)"
+  fi
+}
+
+assert_fuse_ffprobe_ok() {
+  local desc="$1" file="$2"
+  check_client_alive || return 1
+  if timeout "$READ_TIMEOUT" ffprobe -hide_banner -v quiet -show_format "$file" >/dev/null 2>&1; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc (ffprobe failed)"
+  fi
+}
+
+# ── FUSE directory assertion ──
+assert_dir_listing() {
+  local desc="$1" dir="$2" expected="$3"
+  check_client_alive || return 1
+  local result; result="$(timeout "$READ_TIMEOUT" ls "$dir" 2>/dev/null || true)"
+  if [[ "$result" == *"$expected"* ]]; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc (directory missing '$expected')"
+  fi
+}
+
+# ── FUSE write assertions ──
+assert_metaflac_write_readback() {
+  local desc="$1" file="$2" tag="$3" value="$4"
+  check_client_alive || return 1
+  metaflac --remove-tag="$tag" --set-tag="$tag=$value" "$file" 2>/dev/null
+  local result; result="$(metaflac --show-tag="$tag" "$file" 2>/dev/null || true)"
+  if [[ "$result" == *"$value"* ]]; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc (write readback mismatch)"
+  fi
+}
+
+assert_mp3_write_readback() {
+  local desc="$1" file="$2" song="$3"
+  check_client_alive || return 1
+  id3v2 --song "$song" "$file" 2>/dev/null
+  local result; result="$(id3v2 --list "$file" 2>/dev/null || true)"
+  if [[ "$result" == *"$song"* ]]; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc (MP3 write readback mismatch)"
+  fi
+}
+
+# ── log assertions ──
+_get_log_file() {
+  ls -t "$LOG_DIR"/musicfs.*.log 2>/dev/null | head -1
+}
+
+assert_log_contains() {
+  local desc="$1" pattern="$2"
+  local log_file; log_file="$(_get_log_file)"
+  if [[ -n "$log_file" ]] && grep -q "$pattern" "$log_file" 2>/dev/null; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc (pattern '$pattern' not in logs)"
+  fi
+}
+
+log_contains_since() {
+  local pattern="$1" from_line="$2"
+  local log_file; log_file="$(_get_log_file)"
+  [[ -n "$log_file" ]] && tail -n +"$((from_line + 1))" "$log_file" 2>/dev/null | grep -q "$pattern"
+}
+
+log_line_count() {
+  local log_file; log_file="$(_get_log_file)"
+  if [[ -n "$log_file" ]]; then
+    wc -l < "$log_file" 2>/dev/null || echo 0
+  else
+    echo 0
+  fi
+}
+
+assert_log_contains_after() {
+  local desc="$1" pattern="$2" marker="$3"
+  local log_file; log_file="$(_get_log_file)"
+  if [[ -z "$log_file" ]]; then
+    _res_fail "$desc (no log file)"
+    return
+  fi
+  # Find line number of the last occurrence of the marker.
+  local marker_line
+  marker_line="$(grep -n "$marker" "$log_file" 2>/dev/null | tail -1 | cut -d: -f1)"
+  if [[ -z "$marker_line" ]]; then
+    _res_fail "$desc (marker '$marker' not in log)"
+    return
+  fi
+  if tail -n +"$((marker_line + 1))" "$log_file" 2>/dev/null | grep -q "$pattern"; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc ('$pattern' not found after marker line $marker_line)"
+  fi
+}
+
+# ── recovery wait ──
+wait_recover() {
+  local file="$1"
+  local timeout="${2:-$RECOVER_TIMEOUT}"
+  local i
+  for ((i = 0; i < timeout; i++)); do
+    check_client_alive || return 1
+    if timeout 5 dd if="$file" of=/dev/null bs=4096 2>/dev/null; then
+      return 0
+    fi
+    sleep 1
+  done
+  return 1
+}
+
+assert_recovered() {
+  local desc="$1" file="$2"
+  if wait_recover "$file"; then
+    _res_pass "$desc"
+  else
+    _res_fail "$desc (did not recover within ${RECOVER_TIMEOUT}s)"
+  fi
+}
+
+# ── scenario isolation ──
+ensure_clean_state() {
+  log "ensuring clean state between scenarios…"
+  # Server must be up
+  if ! wait_health 5 2>/dev/null; then
+    server_start
+    wait_health "$HEALTH_TIMEOUT" || die "server won't come up between scenarios"
+  fi
+  # VM must be running (not paused)
+  local state
+  state="$(incus list "$VM_NAME" -f csv -c s 2>/dev/null | head -1)"
+  if [[ "$state" == "FROZEN" ]]; then
+    vm_resume
+    wait_health "$HEALTH_TIMEOUT" || die "server unhealthy after VM resume"
+  fi
+  check_client_alive || die "client died between scenarios"
+  clear_cache
+}

scripts/e2e/run-resilience.sh

@@ -0,0 +1,423 @@
+#!/usr/bin/env bash
+#
+# run-resilience.sh — network stability and reconnection E2E suite for musicfs.
+#
+# Tests client behavior during server outages, network partitions (VM pause),
+# cache resilience, and long-outage watcher recovery.
+#
+# Requires: devenv shell (for ffmpeg, metaflac, id3v2, incus, grpcurl, Postgres)
+#
+# Usage:
+#   scripts/e2e/run-resilience.sh          full suite
+#   KEEP=1 scripts/e2e/run-resilience.sh   keep VM + DB alive 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=""
+VM_IP=""
+
+. "$SCRIPT_DIR/lib.sh"
+. "$SCRIPT_DIR/resilience-helpers.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
+
+  VM_IP="$(vm_ip)"
+  [[ -n "$VM_IP" ]] || die "could not determine VM IP"
+  log "VM IP: $VM_IP"
+
+  log "waiting for health check…"
+  local i
+  for ((i = 0; i < 45; i++)); do
+    if grpcurl -plaintext "$VM_IP:$PORT" grpc.health.v1.Health/Check 2>/dev/null | grep -q SERVING; then
+      log "server healthy"
+      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:-<unknown>}"
+    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"
+
+# ════════════════════════════════════════════════════════════════════════
+# RESILIENCE SCENARIOS
+# ════════════════════════════════════════════════════════════════════════
+
+# ── Scenario 1: Server Stop → Outage Window → Server Start → Recovery ──
+#
+# Disruption: systemctl stop musicfs-server.service
+# Tests: FUSE error handling (EIO for uncached reads), cache fallback,
+#        tonic lazy channel auto-reconnect after clean server shutdown.
+test_server_stop_start() {
+  printf '\n\033[1;36m── scenario 1: server stop/start ──\033[0m\n'
+  clear_cache
+
+  local watch_marker; watch_marker="$(log_line_count)"
+
+  # Baseline: read works
+  assert_fuse_read_ok "baseline — read Alpha (uncached)" "$ALPHA"
+
+  # Cache Alpha by reading it; leave Gamma uncached
+  # (Alpha is now cached from the read above)
+
+  # ── Disruption: stop server ──
+  server_stop
+  assert_health_down "server is down after stop"
+
+  # During outage
+  assert_fuse_read_ok "outage — cached read Alpha succeeds" "$ALPHA"
+  assert_fuse_eio_within "outage — uncached read Gamma fails (EIO)" "$GAMMA"
+  assert_dir_listing "outage — directory listing still works" "$MNT" "E2E Artist A"
+
+  # ── Recovery: start server ──
+  server_start
+  assert_health_up "server recovered after start"
+  assert_fuse_read_ok "recovery — uncached read Gamma succeeds" "$GAMMA"
+  assert_metaflac_write_readback "recovery — FLAC write after outage" "$BETA" "ARTIST" "Recovered Artist"
+
+  # EXPECTED FAILURE: watcher should resubscribe quickly after a clean server
+  # restart. Currently it sleeps a full POLL_FALLBACK_INTERVAL (30s) before
+  # retrying — no exponential backoff or fast-retry on recoverable errors.
+  local resub=0 j
+  for ((j = 0; j < 10; j++)); do
+    if log_contains_since "subscribed to /events" "$watch_marker"; then
+      resub=1
+      break
+    fi
+    sleep 1
+  done
+  if (( resub == 1 )); then
+    _res_pass "watcher resubscribed within 10s of server restart"
+  else
+    _res_fail "watcher resubscribed within 10s of server restart (stuck in 30s retry — no fast reconnect)"
+  fi
+}
+
+# ── Scenario 2: Network Partition via incus pause → Resume ──
+#
+# Disruption: incus pause (freezes VM — TCP connections go silent)
+# Tests: Silent partition behavior. Uncached reads will hang (no tonic
+#        timeout/keepalive). After resume, client should recover.
+#
+# IMPORTANT: The FUSE daemon (fuser) dispatches requests serially on a single
+# thread. An uncached read during a silent partition blocks the daemon in
+# block_on(fetch_file_range()) indefinitely — queuing ALL subsequent FUSE
+# operations. Therefore:
+#   1. Do all non-blocking FUSE ops (dir listing) BEFORE the blocking read.
+#   2. Resume the VM IMMEDIATELY after the blocking read.
+#   3. Only attempt more FUSE ops after the daemon unblocks.
+test_network_partition() {
+  printf '\n\033[1;36m── scenario 2: network partition (incus pause) ──\033[0m\n'
+  ensure_clean_state
+
+  # Baseline
+  assert_fuse_read_ok "baseline — read Alpha" "$ALPHA"
+
+  # ── Disruption: freeze VM ──
+  vm_pause
+  assert_health_down "server unreachable during pause"
+
+  # Dir listing works (in-memory, no network) — must do this BEFORE the
+  # blocking read below, which freezes the single-threaded FUSE daemon.
+  assert_dir_listing "partition — directory listing (pre-block)" "$MNT" "E2E Artist A"
+
+  # EXPECTED FAILURE: with proper tonic timeout/keepalive, this read should
+  # return EIO within ~10s. Instead it HANGS indefinitely because no timeout
+  # is configured — block_on(fetch_file_range()) never returns while the VM
+  # is paused. timeout kills dd at 10s, but the daemon thread stays stuck.
+  assert_fuse_eio_within "partition — uncached read Gamma returns EIO within 10s" "$GAMMA" 10
+
+  # ── Recovery: resume VM immediately ──
+  # After resume, the pending gRPC call completes (server processes the
+  # in-flight request), unblocking the FUSE daemon thread.
+  vm_resume
+  assert_health_up "server recovered after resume"
+  assert_recovered "recovery — client reads after partition" "$GAMMA"
+  assert_fuse_read_ok "recovery — cached read Alpha" "$ALPHA"
+}
+
+# ── Scenario 3: Cache Resilience During Full Outage ──
+#
+# Disruption: systemctl stop (server down for entire test)
+# Tests: Postgres-backed cache provides degraded read-only mode.
+#        All pre-warmed files are fully readable. Writes to cached files
+#        work (local DB path). Writes to uncached files fail gracefully.
+test_cache_resilience() {
+  printf '\n\033[1;36m── scenario 3: cache resilience during outage ──\033[0m\n'
+  ensure_clean_state
+
+  # Pre-warm: read every file so all are in Postgres cache
+  prewarm_all
+
+  # Verify prewarm worked (all files cached)
+  assert_fuse_metaflac_ok "prewarm — Alpha metaflac read" "$ALPHA"
+  assert_fuse_id3v2_ok "prewarm — Gamma id3v2 read" "$GAMMA"
+
+  # ── Disruption: stop server for entire scenario ──
+  server_stop
+  assert_health_down "server down for cache test"
+
+  # Cached reads — all should succeed from Postgres cache
+  assert_fuse_metaflac_ok "outage — cached FLAC metaflac (Alpha)" "$ALPHA"
+  assert_fuse_metaflac_ok "outage — cached FLAC metaflac (Beta)" "$BETA"
+  assert_fuse_id3v2_ok "outage — cached MP3 id3v2 (Gamma)" "$GAMMA"
+  assert_fuse_id3v2_ok "outage — cached MP3 id3v2 (Delta)" "$DELTA"
+  assert_fuse_ffprobe_ok "outage — cached ffprobe (Alpha)" "$ALPHA"
+
+  # Write to cached file — should succeed (write path is local DB only)
+  assert_metaflac_write_readback "outage — write to cached FLAC (Beta)" "$BETA" "ARTIST" "Cache Outage Artist"
+
+  # Directory listing — in-memory, no server needed
+  assert_dir_listing "outage — dir listing works" "$MNT" "E2E Artist B"
+
+  # ── Recovery ──
+  server_start
+  assert_health_up "server recovered"
+  assert_metaflac_write_readback "recovery — write after cache outage" "$BETA" "ARTIST" "Post Outage Artist"
+}
+
+# ── Scenario 4: Long Outage + Watcher Recovery ──
+#
+# Disruption: incus pause for 90s (> 2x POLL_FALLBACK_INTERVAL of 30s)
+# Tests: Watcher retry loop stamina during prolonged partition.
+#        After resume, watcher resubscribes and reconcile succeeds.
+test_long_outage_watcher() {
+  printf '\n\033[1;36m── scenario 4: long outage + watcher recovery ──\033[0m\n'
+  ensure_clean_state
+
+  assert_log_contains "baseline — watcher subscribed" "subscribed to /events"
+
+  local marker; marker="$(log_line_count)"
+
+  vm_pause
+  assert_health_down "server unreachable during long pause"
+
+  # EXPECTED FAILURE: watcher should detect the dead stream within ~30s
+  # (POLL_FALLBACK_INTERVAL). Instead it's stuck on stream.next() for ~70s
+  # until the HTTP/2 transport layer times out — no keepalive is configured
+  # to detect the silent partition sooner.
+  local watcher_pattern="stream error\|subscribe failed\|poll reconcile failed"
+  local detected=0
+  local i
+  for ((i = 0; i < 30; i++)); do
+    check_client_alive || break
+    if log_contains_since "$watcher_pattern" "$marker"; then
+      detected=1
+      break
+    fi
+    sleep 1
+  done
+
+  if (( detected == 1 )); then
+    _res_pass "watcher detected silent partition within 30s"
+  else
+    _res_fail "watcher detected silent partition within 30s (stuck on stream.next — no keepalive)"
+  fi
+
+  assert_dir_listing "long outage — dir listing works" "$MNT" "E2E Artist A"
+  check_client_alive || die "client died during long outage"
+
+  vm_resume
+  assert_health_up "server recovered after long pause"
+  assert_recovered "client reads after long outage" "$ALPHA"
+
+  local resub=0 j
+  for ((j = 0; j < 10; j++)); do
+    if log_contains_since "subscribed to /events" "$marker"; then
+      resub=1
+      break
+    fi
+    sleep 1
+  done
+  if (( resub == 1 )); then
+    _res_pass "watcher resubscribed within 10s of recovery"
+  else
+    _res_fail "watcher resubscribed within 10s of recovery (stuck in 30s retry — no fast reconnect)"
+  fi
+}
+
+# ════════════════════════════════════════════════════════════════════════
+# 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 "=== Resilience 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: resilience scenarios
+  test_server_stop_start
+  test_network_partition
+  test_cache_resilience
+  test_long_outage_watcher
+
+  # Summary
+  printf '\n'
+  summary
+  local rv=$?
+  log "=== Resilience E2E suite done ==="
+  exit $rv
+}
+
+main "$@"

scripts/vm.sh

@@ -1,4 +1,4 @@
-#!/usr/bin/env bash
+#!/usr/bin/env -S nix shell nixpkgs#bash --command bash
 #
 # vm.sh - Incus VM launcher for musicfs-server
 #
@@ -30,6 +30,8 @@
 #   MUSIC_MOUNT   where it appears inside the VM       (default: /music)
 #   LISTEN_PORT   gRPC port (host:vm forwarded)        (default: 50051)
 #   RUST_LOG      tracing filter for the server        (default: info)
+#   READONLY      mount music as read-only (0=writable) (default: 1)
+#   VM_IP         static IPv4 for the VM                (default: auto-detected from incusbr0, gateway.145)
 #
 set -euo pipefail
 
@@ -40,8 +42,15 @@ MUSIC_SOURCE="${MUSIC_SOURCE:-$HOME/Music}"
 MUSIC_MOUNT="${MUSIC_MOUNT:-/music}"
 LISTEN_PORT="${LISTEN_PORT:-50051}"
 RUST_LOG="${RUST_LOG:-info}"
+detect_static_ip() {
+  if [[ -n "${VM_IP:-}" ]]; then echo "$VM_IP"; return; fi
+  local bridge_cidr
+  bridge_cidr="$(incus network get incusbr0 ipv4.address 2>/dev/null || true)"
+  if [[ -z "$bridge_cidr" ]]; then echo ""; return; fi
+  local gateway="${bridge_cidr%%/*}"
+  echo "${gateway%.*}.145"
+}
 
-# Fixed inside the VM; the bundle's interpreter+rpath point here.
 SERVER_DIR="/opt/musicfs"
 LOG_DIR="/var/log/musicfs"
 BIN_NAME="musicfs-server"
@@ -60,7 +69,12 @@ vm_running() { incus list "$VM_NAME" -f csv -c n,s 2>/dev/null | grep -q "$VM_NA
 device_exists() { incus config device show "$VM_NAME" 2>/dev/null | grep -q "^$1:"; }
 
 # incus -c 4 emits "<ip> (<iface>)"; keep only the address.
-vm_ip() { incus list "$VM_NAME" -f csv -c 4 2>/dev/null | head -1 | awk '{print $1}'; }
+vm_ip() {
+  local ip
+  ip="$(incus list "$VM_NAME" -f csv -c 4 2>/dev/null | head -1 | awk '{print $1}')"
+  if [[ -n "$ip" ]]; then echo "$ip"; return; fi
+  detect_static_ip
+}
 
 # ──────────────────────────── build + bundle ───────────────────────────
 # Emits the host path to the prepared bundle dir on stdout.
@@ -120,10 +134,24 @@ create_vm() {
   log "creating VM '$VM_NAME' from $IMAGE ..."
   incus init "$IMAGE" "$VM_NAME" --vm
 
+  local static_ip
+  static_ip="$(detect_static_ip)"
+  if [[ -n "$static_ip" ]]; then
+    log "reserving static IP: $static_ip"
+    incus config device add "$VM_NAME" eth0 nic \
+      network=incusbr0 name=eth0 ipv4.address="$static_ip"
+  fi
+
   if ! device_exists music; then
-    log "sharing music: $MUSIC_SOURCE -> $MUSIC_MOUNT (readonly, virtiofs)"
+    local ro_flag="readonly=true"
+    local mode="read-only"
+    if [[ "$READONLY" == "0" ]]; then
+      ro_flag="readonly=false"
+      mode="read-write"
+    fi
+    log "sharing music: $MUSIC_SOURCE -> $MUSIC_MOUNT ($mode, virtiofs)"
     incus config device add "$VM_NAME" music disk \
-      source="$MUSIC_SOURCE" path="$MUSIC_MOUNT" readonly=true
+      source="$MUSIC_SOURCE" path="$MUSIC_MOUNT" "$ro_flag"
   fi
 
   # No proxy: Incus VM proxies need NAT mode + a *static* instance IP (DHCP is
@@ -255,7 +283,15 @@ print_status() {
   printf '  ipv4:    %s\n' "${ip:-<none>}"
   printf '  image:   %s\n' "$IMAGE"
   printf '  listen:  0.0.0.0:%s (reach via VM bridge IP above)\n' "$LISTEN_PORT"
-  printf '  music:   %s -> %s\n' "$MUSIC_SOURCE" "$MUSIC_MOUNT"
+  local ro_actual="read-only"
+  if vm_exists; then
+    local ro_val
+    ro_val="$(incus config device get "$VM_NAME" music readonly 2>/dev/null || true)"
+    [[ "$ro_val" == "false" ]] && ro_actual="read-write"
+  else
+    [[ "$READONLY" == "0" ]] && ro_actual="read-write"
+  fi
+  printf '  music:   %s -> %s (%s)\n' "$MUSIC_SOURCE" "$MUSIC_MOUNT" "$ro_actual"
   printf '  server:  %s/server\n' "$SERVER_DIR"
   if [ -n "$ip" ]; then
     printf '\n  test:    grpcurl -plaintext %s:%s list\n\n' "$ip" "$LISTEN_PORT"
@@ -273,7 +309,18 @@ usage() {
 }
 
 main() {
-  case "${1:-up}" in
+  local cmd="${1:-up}"
+  shift || true
+  while (( $# )); do
+    case "$1" in
+      --writable) [[ -z "${READONLY:-}" ]] && READONLY=0 ;;
+      --readonly) [[ -z "${READONLY:-}" ]] && READONLY=1 ;;
+      *) die "unknown flag '$1'. Try: $0 --help" ;;
+    esac
+    shift
+  done
+  READONLY="${READONLY:-1}"
+  case "$cmd" in
     up)        cmd_up ;;
     redeploy)  cmd_redeploy ;;
     restart)   cmd_restart ;;
@@ -283,7 +330,7 @@ main() {
     down)      cmd_down ;;
     destroy)   cmd_destroy ;;
     -h|--help|help) usage ;;
-    *) die "unknown command '$1'. Try: $0 --help" ;;
+    *) die "unknown command '$cmd'. Try: $0 --help" ;;
   esac
 }
 

src/music/metadata.rs

@@ -1,6 +1,6 @@
 use symphonia::core::meta::{MetadataRevision, StandardTagKey};
 
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
 pub struct MusicMetadata {
     pub artist: Vec<String>,
     pub album_artist: Option<String>,

src/origins/network/mod.rs

@@ -115,6 +115,7 @@ impl NetworkOrigin {
                 io_err(e)
             })?
             .into_iter()
+            .filter(|m| m.file_type == "file")
             .map(|m| (m.inode as u64, m.hash as u64))
             .collect();
 
@@ -166,59 +167,9 @@ impl NetworkOrigin {
         }
         *self.latest_manifest.write().unwrap() = current_manifest.clone();
 
-        // 5. Convert manifest entries → Items.
-        let dest = self.destination.clone();
-        let mut snapshot = BTreeMap::new();
-        let root_attrs = FileAttrs {
-            size: 0,
-            blocks: 0,
-            atime: SystemTime::UNIX_EPOCH,
-            mtime: SystemTime::UNIX_EPOCH,
-            ctime: SystemTime::UNIX_EPOCH,
-            crtime: SystemTime::UNIX_EPOCH,
-            perm: 0o755,
-            nlink: 2,
-            uid: 0,
-            gid: 0,
-            rdev: 0,
-            blksize: 4096,
-        };
-        snapshot.insert(
-            INodeNo::ROOT,
-            Item::new(
-                INodeNo::ROOT,
-                INodeNo::ROOT,
-                "/".to_string(),
-                dest.clone(),
-                dest.clone(),
-                FileType::Directory,
-                root_attrs,
-                None,
-            ),
-        );
-
-        let source_root = PathBuf::from("/");
-        for (_id, entry) in &current_manifest {
-            let item = manifest_entry_to_item(entry, &source_root, &mut snapshot);
-            snapshot.insert(item.inode, item);
-        }
-
-        // 6. Reconcile with DB exactly like LocalOrigin: insert/update/delete
-        //    rows, then restore music metadata + virtual paths.
-        let db_items: std::collections::HashMap<i64, item_entities::Model> =
-            item_entities::Entity::find()
-                .all(&self.client)
-                .await
-                .map_err(|e| {
-                    error!(error = %e, "network snapshot: db_items DB read failed");
-                    io_err(e)
-                })?
-                .into_iter()
-                .map(|e| (e.inode, e))
-                .collect();
-        sync_items_to_db(&snapshot, &db_items, &self.client).await;
-        restore_music_metadata_from_db(&mut snapshot, &db_items, &self.client).await;
-        restore_virtual_paths(&mut snapshot, &db_items, &source_root);
+        let snapshot =
+            build_snapshot_from_manifest(&current_manifest, &self.destination, &self.client)
+                .await?;
 
         info!(
             changed = response.changed.len(),
@@ -230,6 +181,70 @@ impl NetworkOrigin {
     }
 }
 
+/// Build a complete Items snapshot from the server manifest, sync it to the
+/// DB, and restore music metadata + virtual paths from existing DB rows.
+///
+/// Shared between `snapshot_async` (initial mount) and the watcher's
+/// `reconcile_once` (runtime updates) so both paths produce identical
+/// snapshots and keep the DB in sync.
+pub(crate) async fn build_snapshot_from_manifest(
+    manifest: &BTreeMap<u64, ProtoManifestEntry>,
+    destination: &Path,
+    client: &sea_orm::DatabaseConnection,
+) -> io::Result<BTreeMap<INodeNo, Item>> {
+    let source_root = PathBuf::from("/");
+    let mut snapshot = BTreeMap::new();
+    let root_attrs = FileAttrs {
+        size: 0,
+        blocks: 0,
+        atime: SystemTime::UNIX_EPOCH,
+        mtime: SystemTime::UNIX_EPOCH,
+        ctime: SystemTime::UNIX_EPOCH,
+        crtime: SystemTime::UNIX_EPOCH,
+        perm: 0o755,
+        nlink: 2,
+        uid: 0,
+        gid: 0,
+        rdev: 0,
+        blksize: 4096,
+    };
+    snapshot.insert(
+        INodeNo::ROOT,
+        Item::new(
+            INodeNo::ROOT,
+            INodeNo::ROOT,
+            "/".to_string(),
+            destination.to_path_buf(),
+            destination.to_path_buf(),
+            FileType::Directory,
+            root_attrs,
+            None,
+        ),
+    );
+
+    for (_id, entry) in manifest {
+        let item = manifest_entry_to_item(entry, &source_root, &mut snapshot);
+        snapshot.insert(item.inode, item);
+    }
+
+    let db_items: std::collections::HashMap<i64, item_entities::Model> =
+        item_entities::Entity::find()
+            .all(client)
+            .await
+            .map_err(|e| {
+                error!(error = %e, "build_snapshot_from_manifest: DB read failed");
+                io_err(e)
+            })?
+            .into_iter()
+            .map(|e| (e.inode, e))
+            .collect();
+    sync_items_to_db(&snapshot, &db_items, client).await;
+    restore_music_metadata_from_db(&mut snapshot, &db_items, client).await;
+    restore_virtual_paths(&mut snapshot, &db_items, &source_root);
+
+    return Ok(snapshot);
+}
+
 fn manifest_entry_to_item(
     entry: &ProtoManifestEntry,
     source_root: &Path,

src/origins/network/transport.rs

@@ -1,4 +1,5 @@
 use std::sync::Arc;
+use std::time::Duration;
 
 use anyhow::{Context, Result, anyhow};
 use tokio_stream::StreamExt;
@@ -28,6 +29,11 @@ impl NetworkTransport {
         let endpoint: tonic::transport::Endpoint = url
             .try_into()
             .map_err(|e| anyhow!("invalid server URL: {e}"))?;
+        let endpoint = endpoint
+            .timeout(Duration::from_secs(3))
+            .connect_timeout(Duration::from_secs(5))
+            .http2_keep_alive_interval(Duration::from_secs(10))
+            .keep_alive_timeout(Duration::from_secs(5));
         let channel = endpoint.connect_lazy();
         let client = Arc::new(tokio::sync::Mutex::new(MusicFsClient::new(channel)));
         return Ok(NetworkTransport { client, endpoint });

src/origins/network/watcher.rs

@@ -17,7 +17,8 @@ use crate::origins::{FileWatcher, WatcherHandle};
 use crate::proto::ManifestEntry as ProtoManifestEntry;
 use tracing::{info, warn};
 
-const POLL_FALLBACK_INTERVAL: Duration = Duration::from_secs(30);
+const INITIAL_RETRY_DELAY: Duration = Duration::from_secs(1);
+const MAX_RETRY_DELAY: Duration = Duration::from_secs(30);
 
 pub struct NetworkOriginFileWatcher {
     transport: NetworkTransport,
@@ -46,11 +47,6 @@ impl NetworkOriginFileWatcher {
     }
 }
 
-/// `files` is currently unused at runtime by the network watcher — the
-/// authoritative post-reconcile snapshot is written to the DB, and the next
-/// FUSE `readdir`/`lookup` will reflect it because FuseFs re-locks the map.
-/// 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<std::sync::Mutex<BTreeMap<INodeNo, Item>>>) -> WatcherHandle {
         let runtime_handle = self.runtime_handle.clone();
@@ -84,6 +80,7 @@ struct WatcherState {
 
 impl WatcherState {
     async fn run_loop(self, shutdown: Arc<Notify>) {
+        let mut retry_delay = INITIAL_RETRY_DELAY;
         loop {
             let subscribed = tokio::select! {
                 biased;
@@ -93,6 +90,7 @@ impl WatcherState {
             match subscribed {
                 Ok(mut stream) => {
                     info!("network watcher: subscribed to /events");
+                    retry_delay = INITIAL_RETRY_DELAY;
                     loop {
                         let item = tokio::select! {
                             biased;
@@ -120,8 +118,9 @@ impl WatcherState {
             tokio::select! {
                 biased;
                 _ = shutdown.notified() => return,
-                _ = tokio::time::sleep(POLL_FALLBACK_INTERVAL) => {}
+                _ = tokio::time::sleep(retry_delay) => {}
             }
+            retry_delay = (retry_delay * 2).min(MAX_RETRY_DELAY);
             if let Err(e) = self.reconcile_once().await {
                 warn!(error = %e, "network watcher: poll reconcile failed");
             }
@@ -133,6 +132,7 @@ impl WatcherState {
             .all(&self.client)
             .await?
             .into_iter()
+            .filter(|m| m.file_type == "file")
             .map(|m| (m.inode as u64, m.hash as u64))
             .collect();
 
@@ -156,8 +156,29 @@ impl WatcherState {
         }
         *self.latest_manifest.write().unwrap() = current_manifest.clone();
 
-        let _ = self.destination.clone();
-        let _ = self.files.clone();
+        let new_snapshot =
+            super::build_snapshot_from_manifest(&current_manifest, &self.destination, &self.client)
+                .await?;
+
+        {
+            let mut files = self.files.lock().unwrap();
+            files.retain(|ino, _| new_snapshot.contains_key(ino));
+            for (ino, new_item) in &new_snapshot {
+                match files.get(ino) {
+                    Some(existing) if existing.hash == new_item.hash => {}
+                    _ => {
+                        files.insert(*ino, new_item.clone());
+                    }
+                }
+            }
+        }
+
+        info!(
+            changed = response.changed.len(),
+            deleted = response.deleted.len(),
+            total = current_manifest.len(),
+            "network watcher: reconcile applied"
+        );
 
         return Ok(());
     }

src/server/state.rs

@@ -15,7 +15,7 @@ use tracing::info;
 
 /// Server-side entry for one file. Built once on startup from a directory
 /// scan and refreshed by the watcher on inotify events.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct FileEntry {
     pub abs_path: PathBuf,
     pub rel_path: String,
@@ -37,16 +37,17 @@ impl ServerState {
 
     /// Replace the entire map with a fresh scan of `source`. Used on startup
     /// and on watcher-driven reconciliations.
-    pub fn replace_all(&self, source: &Path) -> io::Result<()> {
+    pub fn replace_all(&self, source: &Path) -> io::Result<bool> {
         let entries = scan_directory(source)?;
-        let count = entries.len();
+        let new_map: HashMap<u64, FileEntry> = entries.into_iter().collect();
         let mut map = self.inner.lock().unwrap();
-        map.clear();
-        for (id, entry) in entries {
-            map.insert(id, entry);
+        if *map == new_map {
+            return Ok(false);
         }
-        info!(count, "server state: scan complete");
-        return Ok(());
+        let count = new_map.len();
+        *map = new_map;
+        info!(count, "server state: scan complete (changed)");
+        return Ok(true);
     }
 
     /// Snapshot the current state into a manifest. Order is by inode ascending
@@ -179,4 +180,81 @@ mod tests {
         state.replace_all(source).unwrap();
         assert_eq!(state.manifest().len(), 0);
     }
+
+    #[test]
+    fn replace_all_returns_true_on_first_scan() {
+        let tmp = tempfile::tempdir().unwrap();
+        let source = tmp.path();
+        fs::write(source.join("a.txt"), b"x").unwrap();
+
+        let state = ServerState::new();
+        assert!(state.replace_all(source).unwrap());
+    }
+
+    #[test]
+    fn replace_all_returns_false_when_unchanged() {
+        let tmp = tempfile::tempdir().unwrap();
+        let source = tmp.path();
+        fs::write(source.join("a.txt"), b"hello").unwrap();
+        fs::write(source.join("b.txt"), b"world").unwrap();
+
+        let state = ServerState::new();
+        state.replace_all(source).unwrap();
+        assert!(!state.replace_all(source).unwrap());
+    }
+
+    #[test]
+    fn replace_all_returns_true_after_file_added() {
+        let tmp = tempfile::tempdir().unwrap();
+        let source = tmp.path();
+        fs::write(source.join("a.txt"), b"x").unwrap();
+
+        let state = ServerState::new();
+        state.replace_all(source).unwrap();
+
+        fs::write(source.join("b.txt"), b"y").unwrap();
+        assert!(state.replace_all(source).unwrap());
+    }
+
+    #[test]
+    fn replace_all_returns_true_after_file_removed() {
+        let tmp = tempfile::tempdir().unwrap();
+        let source = tmp.path();
+        fs::write(source.join("a.txt"), b"x").unwrap();
+        fs::write(source.join("b.txt"), b"y").unwrap();
+
+        let state = ServerState::new();
+        state.replace_all(source).unwrap();
+
+        fs::remove_file(source.join("a.txt")).unwrap();
+        assert!(state.replace_all(source).unwrap());
+    }
+
+    #[test]
+    fn replace_all_returns_true_after_file_renamed() {
+        let tmp = tempfile::tempdir().unwrap();
+        let source = tmp.path();
+        fs::write(source.join("old.txt"), b"x").unwrap();
+
+        let state = ServerState::new();
+        state.replace_all(source).unwrap();
+        assert!(!state.replace_all(source).unwrap());
+
+        fs::rename(source.join("old.txt"), source.join("new.txt")).unwrap();
+        assert!(state.replace_all(source).unwrap());
+    }
+
+    #[test]
+    fn replace_all_returns_true_after_content_modified() {
+        let tmp = tempfile::tempdir().unwrap();
+        let source = tmp.path();
+        fs::write(source.join("a.txt"), b"original").unwrap();
+
+        let state = ServerState::new();
+        state.replace_all(source).unwrap();
+        assert!(!state.replace_all(source).unwrap());
+
+        fs::write(source.join("a.txt"), b"modified content").unwrap();
+        assert!(state.replace_all(source).unwrap());
+    }
 }

src/server/watcher.rs

@@ -2,6 +2,7 @@ use std::{
     path::PathBuf,
     sync::{Arc, Mutex},
     thread,
+    time::Duration,
 };
 
 use notify::{EventKind, RecursiveMode, Watcher};
@@ -10,6 +11,8 @@ use tokio::sync::broadcast;
 use crate::server::state::ServerState;
 use tracing::{error, info};
 
+const POLL_INTERVAL: Duration = Duration::from_secs(15);
+
 /// A change observed by the watcher. Pushed onto the broadcast channel for
 /// `/events` subscribers. The client treats these as wake-ups: correctness
 /// always rests on the subsequent `/manifest` hash diff.
@@ -76,25 +79,41 @@ fn run_watcher_loop(
     }
 
     info!(source = %source.display(), "server watcher: scanning for changes");
-    for res in rx {
-        match res {
-            Ok(event) => match event.kind {
-                EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_) => {
-                    let kind = match event.kind {
-                        EventKind::Create(_) => ChangeKind::Create,
-                        EventKind::Modify(_) => ChangeKind::Modify,
-                        EventKind::Remove(_) => ChangeKind::Remove,
-                        _ => continue,
-                    };
-                    if let Err(e) = state.replace_all(&source) {
-                        error!(error = %e, "server watcher: state refresh failed");
-                        continue;
+    loop {
+        match rx.recv_timeout(POLL_INTERVAL) {
+            Ok(res) => match res {
+                Ok(event) => match event.kind {
+                    EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_) => {
+                        let kind = match event.kind {
+                            EventKind::Create(_) => ChangeKind::Create,
+                            EventKind::Modify(_) => ChangeKind::Modify,
+                            EventKind::Remove(_) => ChangeKind::Remove,
+                            _ => continue,
+                        };
+                        match state.replace_all(&source) {
+                            Ok(true) => {
+                                let _ = events_tx.send(ChangeEvent { kind });
+                            }
+                            Ok(false) => {}
+                            Err(e) => {
+                                error!(error = %e, "server watcher: state refresh failed");
+                            }
+                        }
                     }
-                    let _ = events_tx.send(ChangeEvent { kind });
-                }
-                _ => {}
+                    _ => {}
+                },
+                Err(e) => error!(error = %e, "server watcher: inotify error"),
             },
-            Err(e) => error!(error = %e, "server watcher: inotify error"),
+            Err(std::sync::mpsc::RecvTimeoutError::Timeout) => match state.replace_all(&source) {
+                Ok(true) => {
+                    let _ = events_tx.send(ChangeEvent {
+                        kind: ChangeKind::Modify,
+                    });
+                }
+                Ok(false) => {}
+                Err(e) => error!(error = %e, "server watcher: poll rescan failed"),
+            },
+            Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break,
         }
     }
 }