From 0e50d8a2a6e021319b0e2dd068ebaa7210d43b38 Mon Sep 17 00:00:00 2001 From: Alexander Date: Wed, 1 Jul 2026 14:27:43 +0200 Subject: [PATCH] Fix reconcile with remote --- scripts/vm.sh | 63 ++++++++++++++--- src/origins/network/mod.rs | 120 ++++++++++++++++++--------------- src/origins/network/watcher.rs | 30 +++++++-- src/server/watcher.rs | 49 +++++++++----- 4 files changed, 177 insertions(+), 85 deletions(-) diff --git a/scripts/vm.sh b/scripts/vm.sh index 749cce5..fefbcf7 100755 --- a/scripts/vm.sh +++ b/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 " ()"; 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:-}" 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 } diff --git a/src/origins/network/mod.rs b/src/origins/network/mod.rs index 0266813..fb1fd62 100644 --- a/src/origins/network/mod.rs +++ b/src/origins/network/mod.rs @@ -166,59 +166,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 ¤t_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 = - 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(¤t_manifest, &self.destination, &self.client) + .await?; info!( changed = response.changed.len(), @@ -230,6 +180,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, + destination: &Path, + client: &sea_orm::DatabaseConnection, +) -> io::Result> { + 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 = + 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, diff --git a/src/origins/network/watcher.rs b/src/origins/network/watcher.rs index 0a079ba..66b7ba6 100644 --- a/src/origins/network/watcher.rs +++ b/src/origins/network/watcher.rs @@ -47,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>>) -> WatcherHandle { let runtime_handle = self.runtime_handle.clone(); @@ -160,8 +155,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(¤t_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(()); } diff --git a/src/server/watcher.rs b/src/server/watcher.rs index b5579e3..dc17f3d 100644 --- a/src/server/watcher.rs +++ b/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,37 @@ 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, + }; + if let Err(e) = state.replace_all(&source) { + error!(error = %e, "server watcher: state refresh failed"); + continue; + } + let _ = events_tx.send(ChangeEvent { kind }); } - 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) => { + if let Err(e) = state.replace_all(&source) { + error!(error = %e, "server watcher: poll rescan failed"); + continue; + } + let _ = events_tx.send(ChangeEvent { + kind: ChangeKind::Modify, + }); + } + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => break, } } }