diff --git a/Dockerfile b/Dockerfile index 9d51679..14640e2 100644 --- a/Dockerfile +++ b/Dockerfile @@ -7,7 +7,7 @@ # - meta-core Go binary (HTTP API, Mount API, Watcher) # - nginx (dashboard, WebDAV, API proxy) # - Redis (managed by supervisord on leader) -# - rclone (remote mount daemon) +# - rclone (single mount executor — SMB goes through rclone's :smb: backend) # - supervisord (process management - runs on leader only) # # Architecture: @@ -67,27 +67,36 @@ LABEL org.opencontainers.image.source=https://github.com/worph/meta-core LABEL org.opencontainers.image.description="MetaMesh meta-core - centralized data layer with leader election, Redis, WebDAV, mounts" LABEL org.opencontainers.image.licenses=MIT -# Install runtime dependencies +# Pinned rclone release. Alpine's package can lag the latest by several +# minor versions and we depend on VFS cache fixes shipped post-1.65. +ARG RCLONE_VERSION=1.73.5 + +# Install runtime dependencies. Native NFS/CIFS clients intentionally absent — +# all remote mounts go through rclone now. RUN apk add --no-cache \ - # Core services redis \ nginx \ nginx-mod-http-dav-ext \ supervisor \ - rclone \ fuse3 \ - # Mount utilities - nfs-utils \ - cifs-utils \ - # Tools bash \ curl \ python3 \ ca-certificates \ apache2-utils \ - # Process utilities procps +# Install rclone from the official binary release (pinned). +RUN set -eux; \ + apk add --no-cache --virtual .rclone-deps unzip; \ + cd /tmp; \ + curl -fsSL "https://downloads.rclone.org/v${RCLONE_VERSION}/rclone-v${RCLONE_VERSION}-linux-amd64.zip" -o rclone.zip; \ + unzip -q rclone.zip; \ + install -m 0755 "rclone-v${RCLONE_VERSION}-linux-amd64/rclone" /usr/local/bin/rclone; \ + rm -rf rclone.zip "rclone-v${RCLONE_VERSION}-linux-amd64"; \ + apk del .rclone-deps; \ + /usr/local/bin/rclone version | head -2 + # Copy Go binary from builder COPY --from=go-builder /build/meta-core /usr/local/bin/meta-core RUN chmod +x /usr/local/bin/meta-core diff --git a/dashboard/src/pages/Mounts.tsx b/dashboard/src/pages/Mounts.tsx index 787e2e2..a8e6b51 100644 --- a/dashboard/src/pages/Mounts.tsx +++ b/dashboard/src/pages/Mounts.tsx @@ -1,19 +1,47 @@ import { useEffect, useState } from 'react'; +interface MountIOStats { + source: 'rclone'; + bytesRead: number; + bytesWritten?: number; + readBps: number; + writeBps?: number; + readOps?: number; + writeOps?: number; + readIops?: number; + writeIops?: number; + transfersPerSec?: number; + lastSampleAt: number; + intervalMs?: number; +} + interface Mount { id: string; name: string; - type: 'nfs' | 'smb' | 'rclone'; + type: 'smb' | 'rclone'; enabled: boolean; desiredMounted: boolean; mountPath: string; mounted: boolean; error?: string; lastChecked: number; + // VFS cache configuration applied at mount time. Empty string means "use + // server default" (50G / 24h / 5m). + cacheMaxSize?: string; + cacheMaxAge?: string; + dirCacheTime?: string; pollingEnabled?: boolean; pollingIntervalMs?: number; pollingActive?: boolean; lastPolledScan?: number; + // Adaptive scan progress (set by the meta-core poller) + scanning?: boolean; + currentScanStartedAt?: number; // ms; only while scanning + lastScanDurationMs?: number; + nextScanAt?: number; // ms; planned start of next adaptive scan + // Live IO — populated by StatsPoller every ~5s. Daemon-global from rclone, + // so the same numbers appear on every mount card. + ioStats?: MountIOStats; } interface RcloneRemote { @@ -43,6 +71,137 @@ const dangerButtonStyle: React.CSSProperties = { background: '#7f1d1d', }; +// Format byte counts/rates with K/M/G/T prefix (decimal — matches what users +// expect for network speeds; binary prefixes feel pedantic on a "MB/s" +// readout). Returns e.g. "12.4 MB", "850 KB", "2.3 GB". +function formatBytes(n: number): string { + if (!isFinite(n) || n < 0) return '—'; + const units = ['B', 'KB', 'MB', 'GB', 'TB']; + let i = 0; + let v = n; + while (v >= 1000 && i < units.length - 1) { + v /= 1000; + i++; + } + return v >= 100 || i === 0 ? `${v.toFixed(0)} ${units[i]}` : `${v.toFixed(1)} ${units[i]}`; +} + +function formatRate(bps: number): string { + return `${formatBytes(bps)}/s`; +} + +// Render a humanised duration: 5s, 1m 23s, 2h 14m. Sub-second collapses to 0s +// rather than "0ms" — at the second-tick cadence the UI uses, finer precision +// is just noise. +function formatDuration(ms: number): string { + if (ms < 0 || !isFinite(ms)) return '—'; + const s = Math.floor(ms / 1000); + if (s < 60) return `${s}s`; + const m = Math.floor(s / 60); + const remS = s % 60; + if (m < 60) return remS ? `${m}m ${remS}s` : `${m}m`; + const h = Math.floor(m / 60); + const remM = m % 60; + return remM ? `${h}h ${remM}m` : `${h}h`; +} + +// IOStatsLine renders a single line of live throughput info beneath each +// mount card. Source is always rclone after the rclone-only consolidation, +// so we report file-level transfers/s rather than kernel IOPS. +function IOStatsLine({ stats }: { stats: MountIOStats }) { + const opsLabel = 'transfers/s'; + const opsValue = stats.transfersPerSec; + + const ageMs = Date.now() - stats.lastSampleAt; + const stale = ageMs > 30000; // >30s old means sampler hasn't ticked recently + + return ( +

+ ↓ {formatRate(stats.readBps)} + {opsValue !== undefined && opsValue > 0 && ( + <> · {opsValue.toFixed(opsValue >= 10 ? 0 : 1)} {opsLabel} + )} + <> · {formatBytes(stats.bytesRead)} read total + {stats.bytesWritten !== undefined && stats.bytesWritten > 0 && ( + <> · {formatBytes(stats.bytesWritten)} written + )} +

+ ); +} + +interface ScanProgressProps { + mount: Mount; + now: number; +} + +// Progress bar shows scan ETA when running, an idle row with a countdown to +// the next adaptive scan when waiting. We can't show true % complete (we +// don't know the file count up-front), so the running bar is filled by +// elapsed/lastDuration capped at 95% — once we exceed the previous duration, +// it pins at 95% to signal "taking longer than last time". +function ScanProgress({ mount, now }: ScanProgressProps) { + if (!mount.pollingEnabled) return null; + + const isScanning = !!mount.scanning && !!mount.currentScanStartedAt; + const lastDuration = mount.lastScanDurationMs ?? 0; + + let fill = 0; + let label = ''; + + if (isScanning) { + const elapsed = Math.max(0, now - (mount.currentScanStartedAt ?? now)); + if (lastDuration > 0) { + fill = Math.min(0.95, elapsed / lastDuration); + label = `Scanning… ${formatDuration(elapsed)} elapsed (last took ${formatDuration(lastDuration)})`; + } else { + fill = 0.4; // first scan ever — no ETA available, show indeterminate-ish + label = `Scanning… ${formatDuration(elapsed)} elapsed (first scan, no ETA yet)`; + } + } else { + const sinceLast = mount.lastPolledScan ? now - mount.lastPolledScan : 0; + const untilNext = mount.nextScanAt ? Math.max(0, mount.nextScanAt - now) : 0; + if (lastDuration > 0 && mount.lastPolledScan) { + label = `Last scan ${formatDuration(sinceLast)} ago (took ${formatDuration(lastDuration)})`; + if (mount.nextScanAt) label += ` • next in ${formatDuration(untilNext)}`; + } else if (mount.nextScanAt) { + label = `Waiting — next scan in ${formatDuration(untilNext)}`; + } else { + label = 'Polling enabled — waiting for first scan'; + } + } + + return ( +
+
+
+
+

{label}

+
+ ); +} + export default function Mounts() { const [mounts, setMounts] = useState([]); const [remotes, setRemotes] = useState([]); @@ -50,19 +209,23 @@ export default function Mounts() { const [showForm, setShowForm] = useState(false); const [formData, setFormData] = useState({ name: '', - type: 'nfs' as 'nfs' | 'smb' | 'rclone', - nfsServer: '', - nfsPath: '', + type: 'smb' as 'smb' | 'rclone', smbServer: '', smbShare: '', smbUsername: '', smbPassword: '', rcloneRemote: '', rclonePath: '', - pollingEnabled: false, + cacheMaxSize: '', + cacheMaxAge: '', + dirCacheTime: '', + pollingEnabled: true, pollingIntervalMs: 30000, }); const [scanning, setScanning] = useState(null); + // 1Hz ticker so the progress bar's elapsed/countdown updates between API + // refetches (which run every 5s). + const [now, setNow] = useState(() => Date.now()); const fetchMounts = async () => { try { @@ -94,7 +257,11 @@ export default function Mounts() { fetchMounts(); fetchRemotes(); const interval = setInterval(fetchMounts, 5000); - return () => clearInterval(interval); + const tick = setInterval(() => setNow(Date.now()), 1000); + return () => { + clearInterval(interval); + clearInterval(tick); + }; }, []); const handleMount = async (id: string) => { @@ -143,10 +310,7 @@ export default function Mounts() { type: formData.type, }; - if (formData.type === 'nfs') { - body.nfsServer = formData.nfsServer; - body.nfsPath = formData.nfsPath; - } else if (formData.type === 'smb') { + if (formData.type === 'smb') { body.smbServer = formData.smbServer; body.smbShare = formData.smbShare; if (formData.smbUsername) body.smbUsername = formData.smbUsername; @@ -156,6 +320,12 @@ export default function Mounts() { body.rclonePath = formData.rclonePath; } + // Cache knobs — only forward non-empty values; the server applies defaults + // when fields are omitted. + if (formData.cacheMaxSize) body.cacheMaxSize = formData.cacheMaxSize; + if (formData.cacheMaxAge) body.cacheMaxAge = formData.cacheMaxAge; + if (formData.dirCacheTime) body.dirCacheTime = formData.dirCacheTime; + // Add polling configuration body.pollingEnabled = formData.pollingEnabled; if (formData.pollingEnabled && formData.pollingIntervalMs >= 5000) { @@ -171,16 +341,17 @@ export default function Mounts() { setShowForm(false); setFormData({ name: '', - type: 'nfs', - nfsServer: '', - nfsPath: '', + type: 'smb', smbServer: '', smbShare: '', smbUsername: '', smbPassword: '', rcloneRemote: '', rclonePath: '', - pollingEnabled: false, + cacheMaxSize: '', + cacheMaxAge: '', + dirCacheTime: '', + pollingEnabled: true, pollingIntervalMs: 30000, }); fetchMounts(); @@ -231,32 +402,15 @@ export default function Mounts() { - {formData.type === 'nfs' && ( - <> - setFormData({ ...formData, nfsServer: e.target.value })} - required - /> - setFormData({ ...formData, nfsPath: e.target.value })} - required - /> - - )} - {formData.type === 'smb' && ( <> )} +
+

+ VFS cache (leave blank for defaults: 50G / 24h / 5m) +

+ setFormData({ ...formData, cacheMaxSize: e.target.value })} + /> + setFormData({ ...formData, cacheMaxAge: e.target.value })} + /> + setFormData({ ...formData, dirCacheTime: e.target.value })} + /> +
+
)} + )}
{breadcrumb && breadcrumb.length > 0 && ( @@ -237,6 +247,28 @@ export default function KVApp() { + {snapshotOpen && ( +
setSnapshotOpen(false)} + > +
e.stopPropagation()} + role="dialog" + aria-modal="true" + > + + +
+
+ )} + ); diff --git a/editor/tsconfig.tsbuildinfo b/editor/tsconfig.tsbuildinfo index 321e28b..3ad81f0 100644 --- a/editor/tsconfig.tsbuildinfo +++ b/editor/tsconfig.tsbuildinfo @@ -1 +1 @@ -{"root":["./src/App.tsx","./src/main.tsx","./src/vite-env.d.ts","./src/api/kvApi.ts","./src/api/metadataApi.ts","./src/components/BulkMetadataEditor.tsx","./src/components/DynamicFieldRenderer.tsx","./src/components/DynamicPluginSection.tsx","./src/components/FileList.tsx","./src/components/KVTreeView.tsx","./src/components/LanguageCombobox.tsx","./src/components/MetadataEditor.tsx","./src/components/SearchBar.tsx","./src/kv/FilteredKVTree.tsx","./src/kv/KVApp.tsx","./src/kv/KVBranchEditor.tsx","./src/kv/KVLeafEditor.tsx","./src/kv/KVTree.tsx","./src/kv/SearchFieldChips.tsx","./src/kv/Toast.tsx","./src/kv/typeHeuristics.ts","./src/kv/types.ts","./src/types/index.ts","./src/utils/formatting.ts"],"version":"5.9.3"} \ No newline at end of file +{"root":["./src/App.tsx","./src/main.tsx","./src/vite-env.d.ts","./src/api/kvApi.ts","./src/api/metadataApi.ts","./src/api/snapshotApi.ts","./src/components/BulkMetadataEditor.tsx","./src/components/DynamicFieldRenderer.tsx","./src/components/DynamicPluginSection.tsx","./src/components/FileList.tsx","./src/components/KVTreeView.tsx","./src/components/LanguageCombobox.tsx","./src/components/MetadataEditor.tsx","./src/components/SearchBar.tsx","./src/components/SnapshotPanel.tsx","./src/kv/FilteredKVTree.tsx","./src/kv/KVApp.tsx","./src/kv/KVBranchEditor.tsx","./src/kv/KVLeafEditor.tsx","./src/kv/KVTree.tsx","./src/kv/SearchFieldChips.tsx","./src/kv/Toast.tsx","./src/kv/typeHeuristics.ts","./src/kv/types.ts","./src/types/index.ts","./src/utils/formatting.ts"],"version":"5.9.3"} \ No newline at end of file diff --git a/internal/api/server.go b/internal/api/server.go index 957cdbc..8bca8e4 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -29,6 +29,7 @@ type Server struct { mountsManager *mounts.Manager mountsHandlers *mounts.Handlers mountPoller *mounts.Poller + mountStatsPoller *mounts.StatsPoller watcherDispatcher *watcher.Dispatcher fileWatcher *watcher.Watcher watcherHandlers *watcher.Handlers @@ -71,6 +72,11 @@ func NewServer( log.Printf("[API] Warning: failed to initialize file watcher: %v", err) } else { s.fileWatcher = fileWatcher + // Wire storage so the watcher can persist (path,size,mtime,cid) + // tuples to Redis as it computes midhashes. + if stor != nil { + fileWatcher.SetStorageClient(stor) + } s.watcherHandlers = watcher.NewHandlers(fileWatcher, s.watcherDispatcher) // Create scan function from watcher for mounts scanFunc = fileWatcher.ScanMountPath @@ -88,6 +94,20 @@ func NewServer( s.watchersPoller = watchers.NewPoller(watchersManager, s.fileWatcher, s.watcherDispatcher) // Pass republish callback to handlers (handles nil metaPublisher case) s.watchersHandlers = watchers.NewHandlers(watchersManager, s.watchersPoller, s.RepublishMetadata) + + // Hydrate the in-memory state registry from Redis tuples so + // the first scan tick after boot can skip ComputeMidHash256 + // for files whose (size, mtime) match. + configs := watchersManager.List() + roots := make([]watcher.WatcherRoot, 0, len(configs)) + for _, c := range configs { + roots = append(roots, watcher.WatcherRoot{ID: c.ID, Root: c.Path}) + } + if loaded, skipped, err := s.fileWatcher.HydrateStateFromStorage(roots); err != nil { + log.Printf("[API] Warning: failed to hydrate watcher state: %v", err) + } else { + log.Printf("[API] Hydrated watcher state from Redis: %d files loaded, %d skipped (no matching watcher root)", loaded, skipped) + } } } } @@ -103,6 +123,12 @@ func NewServer( // Create and configure mount poller s.mountPoller = mounts.NewPoller(mountsManager, scanFunc) mountsManager.SetPoller(s.mountPoller) + + // IO stats poller — queries the local rclone RC API every few seconds + // and exposes live throughput per mount via the same status JSON. + // Stats are daemon-global (rclone limitation); see stats_poller.go. + s.mountStatsPoller = mounts.NewStatsPoller(mountsManager) + mountsManager.SetStatsPoller(s.mountStatsPoller) } // Initialize WebDAV handler (caching is handled by nginx proxy_cache) @@ -126,6 +152,11 @@ func (s *Server) setupRoutes() { s.router.HandleFunc("/api/metadata/search", s.handleSearchMetadata).Methods("POST") s.router.HandleFunc("/api/metadata/batch", s.handleBatchUpdate).Methods("POST") s.router.HandleFunc("/api/metadata/clear", s.handleClearMetadata).Methods("POST") + + // Snapshot (export / import / wipe) + s.router.HandleFunc("/api/snapshot/export", s.handleSnapshotExport).Methods("GET") + s.router.HandleFunc("/api/snapshot/import", s.handleSnapshotImport).Methods("POST") + s.router.HandleFunc("/api/snapshot/wipe", s.handleSnapshotWipe).Methods("POST") s.router.HandleFunc("/api/metadata/{hashId}/property", s.handleMetadataGetProperty).Methods("GET") s.router.HandleFunc("/api/metadata/{hashId}/property", s.handleMetadataUpdateProperty).Methods("PUT") s.router.HandleFunc("/api/metadata/{hashId}", s.handleGetMetadataByHashId).Methods("GET") @@ -240,6 +271,13 @@ func (s *Server) Start() error { } } + // Start IO stats poller alongside the mount poller. Cheap (one /proc read + // per CIFS/NFS mount per cycle, plus one RC call shared by all rclone + // mounts), so always-on rather than gated by config. + if s.mountStatsPoller != nil { + s.mountStatsPoller.Start() + } + // Start meta publisher (if storage is connected) // Publishes keyspace notifications to meta:events stream if s.storage != nil && s.storage.IsConnected() { @@ -278,6 +316,11 @@ func (s *Server) Stop() error { } } + // Stop IO stats poller + if s.mountStatsPoller != nil { + s.mountStatsPoller.Stop() + } + // Stop meta publisher if s.metaPublisher != nil { if err := s.metaPublisher.Stop(); err != nil { diff --git a/internal/api/snapshot.go b/internal/api/snapshot.go new file mode 100644 index 0000000..9bd962a --- /dev/null +++ b/internal/api/snapshot.go @@ -0,0 +1,167 @@ +package api + +import ( + "bytes" + "fmt" + "io" + "net/http" + "path/filepath" + "strings" + "time" + + "github.com/metazla/meta-core/internal/snapshot" +) + +// Bound the upload size so a malicious client can't OOM the server. +// 2 GiB is well above any realistic export. +const maxImportBytes = 2 << 30 + +// pluginOutputDir is where plugin-produced artifacts live on disk +// (subtitles, posters, etc). Mounted from the host into the meta-core +// container at $FILES_PATH/plugin. +func (s *Server) pluginOutputDir() string { + return filepath.Join(s.config.FilesPath, "plugin") +} + +// handleSnapshotExport streams a ZIP snapshot of all metadata. +// GET /api/snapshot/export?include=hash-index,plugin-output +func (s *Server) handleSnapshotExport(w http.ResponseWriter, r *http.Request) { + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + // The server has a 30s WriteTimeout (server.go:222). Long exports + // blow past that, leaving a truncated ZIP without a central directory. + // Clear the deadline for this handler so the export can run as long as + // needed. ResponseController is the standard way to do this. + rc := http.NewResponseController(w) + _ = rc.SetWriteDeadline(time.Time{}) + + include := splitCSVParam(r.URL.Query().Get("include")) + opts := snapshot.ExportOptions{ + IncludePluginOutput: include["plugin-output"], + PluginOutputDir: s.pluginOutputDir(), + } + + filename := fmt.Sprintf("metamesh-snapshot-%s.zip", time.Now().UTC().Format("20060102-150405")) + w.Header().Set("Content-Type", "application/zip") + w.Header().Set("Content-Disposition", fmt.Sprintf(`attachment; filename="%s"`, filename)) + // We stream — no Content-Length. + + exporter := snapshot.NewExporter(s.storage) + if _, err := exporter.Export(w, opts); err != nil { + // Headers may already be sent; log and bail. + // Client will see a truncated zip, which is the best signal we can give. + fmt.Printf("[Snapshot] export failed: %v\n", err) + return + } +} + +// handleSnapshotImport applies an uploaded ZIP snapshot. +// POST /api/snapshot/import?mode=replace|merge&conflict=mine|source&dry_run=true +// Body: application/zip (raw) OR multipart/form-data with field "snapshot" +func (s *Server) handleSnapshotImport(w http.ResponseWriter, r *http.Request) { + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + // Long imports can blow past the server's 30s ReadTimeout/WriteTimeout + // when the snapshot is large. Clear both for this handler. + rc := http.NewResponseController(w) + _ = rc.SetReadDeadline(time.Time{}) + _ = rc.SetWriteDeadline(time.Time{}) + + q := r.URL.Query() + opts := snapshot.ImportOptions{ + Mode: snapshot.Mode(q.Get("mode")), + Conflict: snapshot.Conflict(q.Get("conflict")), + DryRun: strings.EqualFold(q.Get("dry_run"), "true"), + PluginOutputDir: s.pluginOutputDir(), + } + if opts.Mode == "" { + opts.Mode = snapshot.ModeMerge + } + if opts.Mode == snapshot.ModeMerge && opts.Conflict == "" { + opts.Conflict = snapshot.ConflictSource + } + + body, err := readImportBody(r) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + + importer := snapshot.NewImporter(s.storage) + result, err := importer.Import(bytes.NewReader(body), int64(len(body)), opts) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + + writeJSON(w, http.StatusOK, result) +} + +// handleSnapshotWipe clears Redis metadata. +// POST /api/snapshot/wipe?scope=metadata +func (s *Server) handleSnapshotWipe(w http.ResponseWriter, r *http.Request) { + if !s.storage.IsConnected() { + writeError(w, http.StatusServiceUnavailable, "storage not connected") + return + } + + scopes := splitCSVParam(r.URL.Query().Get("scope")) + if len(scopes) == 0 { + writeError(w, http.StatusBadRequest, "scope query param is required (e.g. scope=metadata)") + return + } + scope := snapshot.WipeScope{ + Metadata: scopes["metadata"], + } + if !scope.Metadata { + writeError(w, http.StatusBadRequest, "no supported scope requested; scope=metadata is the only accepted value") + return + } + + wiper := snapshot.NewWiper(s.storage) + res, err := wiper.Wipe(scope) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + writeJSON(w, http.StatusOK, res) +} + +// readImportBody accepts either a raw application/zip body or a multipart +// upload with a field named "snapshot". Buffers up to maxImportBytes. +func readImportBody(r *http.Request) ([]byte, error) { + r.Body = http.MaxBytesReader(nil, r.Body, maxImportBytes) + + ct := r.Header.Get("Content-Type") + if strings.HasPrefix(ct, "multipart/form-data") { + if err := r.ParseMultipartForm(32 << 20); err != nil { + return nil, fmt.Errorf("parse multipart: %w", err) + } + f, _, err := r.FormFile("snapshot") + if err != nil { + return nil, fmt.Errorf("missing 'snapshot' file field: %w", err) + } + defer f.Close() + return io.ReadAll(f) + } + + return io.ReadAll(r.Body) +} + +// splitCSVParam parses "a,b,c" into a set. Empty values yield an empty set. +func splitCSVParam(v string) map[string]bool { + out := map[string]bool{} + for _, part := range strings.Split(v, ",") { + p := strings.TrimSpace(part) + if p != "" { + out[p] = true + } + } + return out +} diff --git a/internal/mounts/manager.go b/internal/mounts/manager.go index 03145db..8fe7cdb 100644 --- a/internal/mounts/manager.go +++ b/internal/mounts/manager.go @@ -19,11 +19,12 @@ import ( // Manager handles mount configuration and status type Manager struct { - config *config.Config - mu sync.RWMutex - filesPath string - mountsDir string - poller *Poller + config *config.Config + mu sync.RWMutex + filesPath string + mountsDir string + poller *Poller + statsPoller *StatsPoller } // NewManager creates a new mount manager @@ -39,14 +40,55 @@ func NewManager(cfg *config.Config) (*Manager, error) { return nil, fmt.Errorf("failed to create mount directories: %w", err) } + // One-shot migration for legacy NFS entries — disable rather than fail to + // start, so operators can fix their config at their own pace. + if err := m.migrateLegacyMounts(); err != nil { + log.Printf("[Mounts] legacy migration warning: %v", err) + } + return m, nil } +// migrateLegacyMounts disables any mount entries with type="nfs". The native +// NFS handler was removed when the project consolidated on rclone; an NFS +// entry left enabled would loop the watcher with mount errors. +func (m *Manager) migrateLegacyMounts() error { + mountsFile, err := m.readConfig() + if err != nil { + return err + } + + dirty := false + for i := range mountsFile.Mounts { + mnt := &mountsFile.Mounts[i] + if string(mnt.Type) == "nfs" { + if mnt.Enabled || mnt.DesiredMounted { + log.Printf("[Mounts] WARNING: disabling legacy NFS mount %q (%s) — NFS support was removed; please re-add via SMB or a pre-configured rclone remote", mnt.Name, mnt.ID) + mnt.Enabled = false + mnt.DesiredMounted = false + dirty = true + } + } + } + + if dirty { + return m.writeConfig(mountsFile) + } + return nil +} + // SetPoller sets the poller for the manager func (m *Manager) SetPoller(poller *Poller) { m.poller = poller } +// SetStatsPoller sets the IO stats poller for the manager. ListMounts and +// GetMount will merge the latest sampled stats into MountStatus.IOStats when +// set; remains nil-safe when stats are disabled. +func (m *Manager) SetStatsPoller(sp *StatsPoller) { + m.statsPoller = sp +} + // ensureDirs creates required directories func (m *Manager) ensureDirs() error { dirs := []string{ @@ -197,6 +239,15 @@ func (m *Manager) ListMounts() ([]MountStatus, error) { pollingStatus := m.poller.GetPollingStatus(statuses[i].ID) statuses[i].PollingActive = pollingStatus.Active statuses[i].LastPolledScan = pollingStatus.LastScan + statuses[i].Scanning = pollingStatus.Scanning + statuses[i].CurrentScanStartedAt = pollingStatus.CurrentScanStartedAt + statuses[i].LastScanDurationMs = pollingStatus.LastScanDurationMs + statuses[i].NextScanAt = pollingStatus.NextScanAt + } + } + if m.statsPoller != nil { + for i := range statuses { + statuses[i].IOStats = m.statsPoller.GetStats(statuses[i].ID) } } @@ -251,6 +302,13 @@ func (m *Manager) GetMount(id string) (*MountStatus, error) { pollingStatus := m.poller.GetPollingStatus(mount.ID) status.PollingActive = pollingStatus.Active status.LastPolledScan = pollingStatus.LastScan + status.Scanning = pollingStatus.Scanning + status.CurrentScanStartedAt = pollingStatus.CurrentScanStartedAt + status.LastScanDurationMs = pollingStatus.LastScanDurationMs + status.NextScanAt = pollingStatus.NextScanAt + } + if m.statsPoller != nil { + status.IOStats = m.statsPoller.GetStats(mount.ID) } return status, nil @@ -266,16 +324,19 @@ func (m *Manager) CreateMount(req *CreateMountRequest) (*MountStatus, error) { return nil, fmt.Errorf("mount name is required") } - if req.Type != MountTypeNFS && req.Type != MountTypeSMB && req.Type != MountTypeRclone { - return nil, fmt.Errorf("valid mount type (nfs, smb, rclone) is required") + // "nfs" used to be valid; the native handler is gone. Reject loudly so the + // caller knows their integration needs to migrate to a Samba re-export or + // switch to a supported protocol. + if req.Type == MountType("nfs") { + return nil, fmt.Errorf("NFS mounts are no longer supported — please use SMB or a pre-configured rclone remote") + } + + if req.Type != MountTypeSMB && req.Type != MountTypeRclone { + return nil, fmt.Errorf("valid mount type (smb, rclone) is required") } // Validate type-specific fields switch req.Type { - case MountTypeNFS: - if req.NFSServer == "" || req.NFSPath == "" { - return nil, fmt.Errorf("NFS server and path are required") - } case MountTypeSMB: if req.SMBServer == "" || req.SMBShare == "" { return nil, fmt.Errorf("SMB server and share are required") @@ -309,8 +370,11 @@ func (m *Manager) CreateMount(req *CreateMountRequest) (*MountStatus, error) { enabled = *req.Enabled } - // Default polling settings - pollingEnabled := false + // Polling defaults on for both supported types — neither SMB-via-rclone nor + // generic rclone delivers kernel inotify, so a periodic re-scan is the only + // way to pick up upstream changes. Explicit PollingEnabled in the request + // wins. + pollingEnabled := true if req.PollingEnabled != nil { pollingEnabled = *req.PollingEnabled } @@ -326,16 +390,15 @@ func (m *Manager) CreateMount(req *CreateMountRequest) (*MountStatus, error) { Enabled: enabled, DesiredMounted: enabled, // Auto-mount if enabled MountPath: mountPath, - Options: req.Options, + CacheMaxSize: req.CacheMaxSize, + CacheMaxAge: req.CacheMaxAge, + DirCacheTime: req.DirCacheTime, PollingEnabled: pollingEnabled, PollingIntervalMs: pollingIntervalMs, } // Type-specific fields switch req.Type { - case MountTypeNFS: - mount.NFSServer = req.NFSServer - mount.NFSPath = req.NFSPath case MountTypeSMB: mount.SMBServer = req.SMBServer mount.SMBShare = req.SMBShare @@ -414,6 +477,24 @@ func (m *Manager) UpdateMount(id string, updates map[string]interface{}) (*Mount } } + // VFS cache knobs — strings round-tripped to rclone unchanged, so we + // don't validate format here; rclone rejects bad values at mount time. + if val, ok := updates["cacheMaxSize"]; ok { + if s, ok := val.(string); ok { + mount.CacheMaxSize = s + } + } + if val, ok := updates["cacheMaxAge"]; ok { + if s, ok := val.(string); ok { + mount.CacheMaxAge = s + } + } + if val, ok := updates["dirCacheTime"]; ok { + if s, ok := val.(string); ok { + mount.DirCacheTime = s + } + } + if err := m.writeConfig(mountsFile); err != nil { return nil, err } diff --git a/internal/mounts/poller.go b/internal/mounts/poller.go index 00d7f1e..4870cd0 100644 --- a/internal/mounts/poller.go +++ b/internal/mounts/poller.go @@ -9,6 +9,25 @@ import ( // ScanFunc is a function type for scanning a mount path type ScanFunc func(path string) (int, error) +// IdleFactor controls the adaptive scheduler's duty cycle. The next scan is +// delayed by max(pollingIntervalMs, lastScanDurationMs * IdleFactor), which +// pins steady-state work to roughly 1/(IdleFactor+1) of wall-clock time. With +// IdleFactor=9 a 60s scan triggers a 9-min rest (≈10% duty cycle), keeping +// large remote shares (CORN, etc.) from monopolising the network/CPU. +const IdleFactor = 9 + +// InitialScanDelay is how long a freshly-started poller waits before its first +// scan. Avoids hammering the system at container startup when many mounts come +// up at once. +const InitialScanDelay = 2 * time.Second + +// ReconcileInterval drives the periodic resync loop that keeps poller +// goroutines aligned with mount state. NotifyMountChanged covers config edits, +// but the kernel mount itself is brought up asynchronously by mount-watcher.sh +// — without periodic reconciliation, the poller would never observe the +// transition from `mounted=false` to `mounted=true` and never start scanning. +const ReconcileInterval = 10 * time.Second + // mountPoller manages polling for a single mount type mountPoller struct { mountID string @@ -19,8 +38,12 @@ type mountPoller struct { // PollingStatus holds the polling state for a mount type PollingStatus struct { - Active bool `json:"active"` - LastScan int64 `json:"lastScan,omitempty"` + Active bool `json:"active"` + LastScan int64 `json:"lastScan,omitempty"` // ms; completion timestamp + Scanning bool `json:"scanning,omitempty"` // true while a scan is in flight + CurrentScanStartedAt int64 `json:"currentScanStartedAt,omitempty"` // ms; start of in-flight scan + LastScanDurationMs int64 `json:"lastScanDurationMs,omitempty"` // duration of the previous scan + NextScanAt int64 `json:"nextScanAt,omitempty"` // ms; planned start of next adaptive scan } // Poller manages per-mount polling goroutines @@ -31,8 +54,9 @@ type Poller struct { pollers map[string]*mountPoller // key: mount ID status map[string]*PollingStatus - running bool - mu sync.RWMutex + running bool + reconcileStop chan struct{} + mu sync.RWMutex } // NewPoller creates a new mount poller @@ -54,15 +78,35 @@ func (p *Poller) Start() error { return nil } p.running = true + p.reconcileStop = make(chan struct{}) log.Println("[Poller] Starting mount polling service") // Initial sync with mount configurations go p.syncWithMounts() + // Periodic reconciliation — picks up mount-watcher.sh transitions + // (mounted/unmounted) that NotifyMountChanged doesn't observe. + go p.reconcileLoop(p.reconcileStop) + return nil } +// reconcileLoop runs syncWithMounts on a fixed cadence so the poller catches +// mount state changes initiated outside of CreateMount/UpdateMount/etc. +func (p *Poller) reconcileLoop(stop <-chan struct{}) { + ticker := time.NewTicker(ReconcileInterval) + defer ticker.Stop() + for { + select { + case <-stop: + return + case <-ticker.C: + p.syncWithMounts() + } + } +} + // Stop stops all polling goroutines func (p *Poller) Stop() error { p.mu.Lock() @@ -75,6 +119,11 @@ func (p *Poller) Stop() error { log.Println("[Poller] Stopping mount polling service") + if p.reconcileStop != nil { + close(p.reconcileStop) + p.reconcileStop = nil + } + // Stop all pollers for id, poller := range p.pollers { p.stopPoller(poller) @@ -127,7 +176,7 @@ func (p *Poller) syncWithMounts() { existingPoller, exists := p.pollers[mount.ID] if shouldPoll { - // Calculate interval + // Calculate interval (used as the floor for adaptive scheduling) interval := mount.PollingIntervalMs if interval < MinPollingIntervalMs { interval = DefaultPollingIntervalMs @@ -135,7 +184,7 @@ func (p *Poller) syncWithMounts() { if !exists { // Start new poller - log.Printf("[Poller] Starting poller for mount %s (interval: %dms)", mount.Name, interval) + log.Printf("[Poller] Starting poller for mount %s (interval floor: %dms, idle factor: %dx)", mount.Name, interval, IdleFactor) p.startPollerForMount(mount.ID, mount.MountPath, interval) } } else if exists { @@ -156,7 +205,13 @@ func (p *Poller) syncWithMounts() { } } -// startPollerForMount starts a polling goroutine for a mount +// startPollerForMount starts the adaptive polling goroutine for a mount. +// +// The loop runs scans sequentially: scan → measure duration → sleep +// max(intervalMs, duration*IdleFactor) → repeat. This makes scans +// self-debouncing (no overlap) and bounds steady-state work to ~10% duty +// cycle on slow shares, while keeping fresh checks on small mounts via the +// configured interval floor. func (p *Poller) startPollerForMount(mountID, mountPath string, intervalMs int) { poller := &mountPoller{ mountID: mountID, @@ -170,17 +225,40 @@ func (p *Poller) startPollerForMount(mountID, mountPath string, intervalMs int) } p.status[mountID].Active = true + floor := time.Duration(intervalMs) * time.Millisecond + go func() { - ticker := time.NewTicker(time.Duration(intervalMs) * time.Millisecond) - defer ticker.Stop() + // Initial delay so a freshly-mounted share doesn't get hammered the + // instant the poller starts. + nextDelay := InitialScanDelay + p.setNextScanAt(mountID, time.Now().Add(nextDelay)) for { select { case <-poller.stopChan: return - case <-ticker.C: - p.runPollScan(mountID, mountPath) + case <-time.After(nextDelay): } + + // Re-check stop after the wake-up to avoid running a scan during shutdown. + select { + case <-poller.stopChan: + return + default: + } + + duration, err := p.executeScan(mountID, mountPath) + if err != nil { + log.Printf("[Poller] Scan failed for mount %s: %v", mountID, err) + } + + // Adaptive backoff: idle for IdleFactor × last duration, with the + // configured interval as a floor so tiny mounts still scan often. + nextDelay = duration * IdleFactor + if nextDelay < floor { + nextDelay = floor + } + p.setNextScanAt(mountID, time.Now().Add(nextDelay)) } }() } @@ -199,54 +277,109 @@ func (p *Poller) stopPoller(poller *mountPoller) { // Update status if status, exists := p.status[poller.mountID]; exists { status.Active = false + status.NextScanAt = 0 } } -// runPollScan executes a scan for a mount -func (p *Poller) runPollScan(mountID, mountPath string) { +// executeScan runs scanFunc with timing + status tracking. Both the adaptive +// loop and manual TriggerScan funnel through here so the UI sees a consistent +// "scanning / last duration / last scan" picture regardless of which path +// kicked it off. Returns the scan duration so the caller can compute the next +// adaptive delay. +func (p *Poller) executeScan(mountID, mountPath string) (time.Duration, error) { if p.scanFunc == nil { - return + return 0, nil } + start := time.Now() + startMs := start.UnixMilli() + + p.mu.Lock() + status, exists := p.status[mountID] + if !exists { + status = &PollingStatus{} + p.status[mountID] = status + } + status.Scanning = true + status.CurrentScanStartedAt = startMs + p.mu.Unlock() + fileCount, err := p.scanFunc(mountPath) - if err != nil { - log.Printf("[Poller] Scan failed for mount %s: %v", mountID, err) - return - } - // Update status + end := time.Now() + duration := end.Sub(start) + p.mu.Lock() if status, exists := p.status[mountID]; exists { - status.LastScan = NowMS() + status.Scanning = false + status.CurrentScanStartedAt = 0 + status.LastScan = end.UnixMilli() + status.LastScanDurationMs = duration.Milliseconds() } p.mu.Unlock() - log.Printf("[Poller] Poll scan complete for mount %s: %d files", mountID, fileCount) + if err == nil { + log.Printf("[Poller] Poll scan complete for mount %s: %d files in %s", mountID, fileCount, duration.Round(time.Millisecond)) + } + + return duration, err } -// TriggerScan triggers an immediate scan for a mount +// setNextScanAt records the projected start time of the next adaptive scan +// (used purely for the UI countdown — the loop's own timer is authoritative). +func (p *Poller) setNextScanAt(mountID string, at time.Time) { + p.mu.Lock() + defer p.mu.Unlock() + status, exists := p.status[mountID] + if !exists { + status = &PollingStatus{} + p.status[mountID] = status + } + status.NextScanAt = at.UnixMilli() +} + +// runPollScan is kept for backwards compatibility with any external caller — +// new code should go through executeScan via the adaptive loop. +func (p *Poller) runPollScan(mountID, mountPath string) { + _, _ = p.executeScan(mountID, mountPath) +} + +// TriggerScan triggers an immediate scan for a mount (the "Scan Now" button). +// Runs synchronously; concurrent calls with the adaptive loop are possible +// but rare in practice — both paths update the same status atomically. func (p *Poller) TriggerScan(mountID, mountPath string) (int, error) { if p.scanFunc == nil { return 0, nil } + start := time.Now() + startMs := start.UnixMilli() + + p.mu.Lock() + status, exists := p.status[mountID] + if !exists { + status = &PollingStatus{} + p.status[mountID] = status + } + status.Scanning = true + status.CurrentScanStartedAt = startMs + p.mu.Unlock() + fileCount, err := p.scanFunc(mountPath) - if err != nil { - return 0, err - } + + end := time.Now() + duration := end.Sub(start) - // Update status p.mu.Lock() if status, exists := p.status[mountID]; exists { - status.LastScan = NowMS() - } else { - p.status[mountID] = &PollingStatus{ - LastScan: NowMS(), - } + status.Scanning = false + status.CurrentScanStartedAt = 0 + status.LastScan = end.UnixMilli() + status.LastScanDurationMs = duration.Milliseconds() } p.mu.Unlock() - return fileCount, nil + return fileCount, err } // GetPollingStatus returns the polling status for a mount @@ -256,8 +389,12 @@ func (p *Poller) GetPollingStatus(mountID string) *PollingStatus { if status, exists := p.status[mountID]; exists { return &PollingStatus{ - Active: status.Active, - LastScan: status.LastScan, + Active: status.Active, + LastScan: status.LastScan, + Scanning: status.Scanning, + CurrentScanStartedAt: status.CurrentScanStartedAt, + LastScanDurationMs: status.LastScanDurationMs, + NextScanAt: status.NextScanAt, } } return &PollingStatus{} diff --git a/internal/mounts/stats.go b/internal/mounts/stats.go new file mode 100644 index 0000000..0839294 --- /dev/null +++ b/internal/mounts/stats.go @@ -0,0 +1,70 @@ +package mounts + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "time" +) + +// StatsSample is a raw cumulative-counter snapshot taken at one instant in +// time. The poller keeps the previous sample per mount and computes rates +// (Bps, transfers/sec) from the delta to the latest sample. +// +// Post-consolidation: rclone is the only sample source, since all mount types +// execute through the local rclone daemon. +type StatsSample struct { + TakenAt time.Time + Source string // "rclone" + BytesRead int64 + BytesWritten int64 + ReadOps int64 + WriteOps int64 + // File-level transfer count from rclone's RC API. Distinct from ReadOps + // (which is unused for rclone — kernel ops counters are not exposed + // through FUSE). + Transfers int64 +} + +// rcloneCoreStats matches the RC API response from POST /core/stats. +type rcloneCoreStats struct { + Bytes int64 `json:"bytes"` + TotalBytes int64 `json:"totalBytes"` + Transfers int64 `json:"transfers"` + TotalTransfers int64 `json:"totalTransfers"` +} + +// SampleRclone queries the local rclone RC API for the cumulative bytes + +// transfers counters. The numbers are global across all active mounts on this +// host — rclone tracks stats per daemon, not per mount. The poller reuses one +// sample for every mounted entry on a tick. +func SampleRclone(ctx context.Context) (*StatsSample, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, "http://127.0.0.1:5572/core/stats", nil) + if err != nil { + return nil, fmt.Errorf("build rclone stats request: %w", err) + } + req.SetBasicAuth("admin", "admin") + + client := &http.Client{Timeout: 3 * time.Second} + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("call rclone /core/stats: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("rclone /core/stats returned %d", resp.StatusCode) + } + + var stats rcloneCoreStats + if err := json.NewDecoder(resp.Body).Decode(&stats); err != nil { + return nil, fmt.Errorf("decode rclone stats: %w", err) + } + + return &StatsSample{ + Source: "rclone", + TakenAt: time.Now(), + BytesRead: stats.Bytes, + Transfers: stats.Transfers, + }, nil +} diff --git a/internal/mounts/stats_poller.go b/internal/mounts/stats_poller.go new file mode 100644 index 0000000..527fbf7 --- /dev/null +++ b/internal/mounts/stats_poller.go @@ -0,0 +1,185 @@ +package mounts + +import ( + "context" + "log" + "sync" + "time" +) + +// StatsInterval drives the IO-stats sampling cadence. 5 s is the usual sweet +// spot — short enough that throughput numbers feel live, long enough that +// /proc reads aren't a load source themselves. +const StatsInterval = 5 * time.Second + +// StatsPoller samples cumulative counters from the local rclone RC API and +// exposes computed rates (bytes/sec, transfers/sec) via GetStats. +// +// Post-consolidation note: every mount type executes through rclone, so +// /proc/fs/cifs/Stats and /proc/self/mountstats are no longer relevant — +// rclone is the only counter source. core/stats is daemon-global across all +// active mounts; the same sample is applied to each mount's status. If +// per-mount byte counters become important, the path forward is rclone stats +// groups (out of scope for this consolidation). +type StatsPoller struct { + manager *Manager + + mu sync.RWMutex + lastSamples map[string]*StatsSample // mountID -> previous raw sample + latestStats map[string]*MountIOStats + stopChan chan struct{} + running bool + pollInterval time.Duration +} + +// NewStatsPoller constructs a StatsPoller backed by the given manager. +func NewStatsPoller(manager *Manager) *StatsPoller { + return &StatsPoller{ + manager: manager, + lastSamples: make(map[string]*StatsSample), + latestStats: make(map[string]*MountIOStats), + pollInterval: StatsInterval, + } +} + +// Start launches the sampling loop. Idempotent. +func (p *StatsPoller) Start() { + p.mu.Lock() + if p.running { + p.mu.Unlock() + return + } + p.running = true + p.stopChan = make(chan struct{}) + stop := p.stopChan + p.mu.Unlock() + + log.Printf("[StatsPoller] Starting IO stats sampler (interval: %s)", p.pollInterval) + go p.loop(stop) +} + +// Stop terminates the sampling loop. Idempotent. +func (p *StatsPoller) Stop() { + p.mu.Lock() + defer p.mu.Unlock() + if !p.running { + return + } + p.running = false + if p.stopChan != nil { + close(p.stopChan) + p.stopChan = nil + } +} + +// GetStats returns the latest computed IO stats for a mount, or nil if no +// stats have been observed yet. +func (p *StatsPoller) GetStats(mountID string) *MountIOStats { + p.mu.RLock() + defer p.mu.RUnlock() + s, ok := p.latestStats[mountID] + if !ok { + return nil + } + // Return a copy so the caller can't mutate our cache. + cp := *s + return &cp +} + +func (p *StatsPoller) loop(stop <-chan struct{}) { + // Take the first sample immediately so rates start populating on the + // second tick instead of waiting two intervals. + p.tick() + + t := time.NewTicker(p.pollInterval) + defer t.Stop() + for { + select { + case <-stop: + return + case <-t.C: + p.tick() + } + } +} + +func (p *StatsPoller) tick() { + mounts, err := p.manager.listMountsWithoutPollingStatus() + if err != nil { + log.Printf("[StatsPoller] listMounts failed: %v", err) + return + } + + // Sample rclone once per tick — its RC API gives global stats across all + // active rclone mounts on this host. Both supported mount types (smb, + // rclone) execute through the same rclone daemon, so the same sample is + // applied to every mounted entry. + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + sample, err := SampleRclone(ctx) + if err != nil { + log.Printf("[StatsPoller] rclone sample failed: %v", err) + return + } + if sample == nil { + return + } + + for _, m := range mounts { + if !m.Mounted { + continue + } + + p.mu.Lock() + prev := p.lastSamples[m.ID] + p.lastSamples[m.ID] = sample + stats := computeStats(prev, sample) + p.latestStats[m.ID] = stats + p.mu.Unlock() + } +} + +// computeStats turns a (previous, current) raw-sample pair into the rate- +// based MountIOStats the API exposes. With no previous sample the rates are +// zero — operators will see cumulative counters but Bps/IOPS only on the +// second poll. +func computeStats(prev, cur *StatsSample) *MountIOStats { + out := &MountIOStats{ + Source: cur.Source, + BytesRead: cur.BytesRead, + BytesWritten: cur.BytesWritten, + ReadOps: cur.ReadOps, + WriteOps: cur.WriteOps, + LastSampleAt: cur.TakenAt.UnixMilli(), + } + if prev == nil { + return out + } + dt := cur.TakenAt.Sub(prev.TakenAt).Seconds() + if dt <= 0 { + return out + } + out.IntervalMs = int64(dt * 1000) + + // Counters can reset to zero (e.g. CIFS module reload, remount). Treat + // negative deltas as "no movement" rather than letting the rate go + // negative or implausibly huge. + if d := cur.BytesRead - prev.BytesRead; d > 0 { + out.ReadBps = float64(d) / dt + } + if d := cur.BytesWritten - prev.BytesWritten; d > 0 { + out.WriteBps = float64(d) / dt + } + if d := cur.ReadOps - prev.ReadOps; d > 0 { + out.ReadIops = float64(d) / dt + } + if d := cur.WriteOps - prev.WriteOps; d > 0 { + out.WriteIops = float64(d) / dt + } + if cur.Source == "rclone" { + if d := cur.Transfers - prev.Transfers; d > 0 { + out.TransfersPerSec = float64(d) / dt + } + } + return out +} diff --git a/internal/mounts/types.go b/internal/mounts/types.go index ac5b26d..9cddeca 100644 --- a/internal/mounts/types.go +++ b/internal/mounts/types.go @@ -2,11 +2,19 @@ package mounts import "time" -// MountType represents the type of mount +// MountType represents the type of mount. +// +// All types execute through the local rclone daemon. "smb" is a UX-level alias +// for rclone's :smb: backend — operators fill in SMB-specific fields and the +// mount-watcher synthesises the rclone remote spec at mount time. "rclone" is +// the generic escape hatch for any pre-defined remote in rclone.conf. +// +// Native NFS/CIFS handlers were removed in favour of the rclone-only path — +// kernel inotify wasn't reliable on those mounts anyway and the watcher polls +// regardless. type MountType string const ( - MountTypeNFS MountType = "nfs" MountTypeSMB MountType = "smb" MountTypeRclone MountType = "rclone" ) @@ -17,6 +25,15 @@ const ( MinPollingIntervalMs = 5000 // 5 seconds ) +// VFS cache defaults — applied per-mount when the caller leaves the field +// blank. Values are strings so they round-trip rclone's parsers (SizeSuffix, +// Duration) untouched. +const ( + DefaultCacheMaxSize = "50G" + DefaultCacheMaxAge = "24h" + DefaultDirCacheTime = "5m" +) + // MountConfig represents a mount configuration type MountConfig struct { // Common fields @@ -26,23 +43,26 @@ type MountConfig struct { Enabled bool `json:"enabled"` DesiredMounted bool `json:"desiredMounted"` MountPath string `json:"mountPath"` - Options string `json:"options,omitempty"` - // NFS-specific fields - NFSServer string `json:"nfsServer,omitempty"` - NFSPath string `json:"nfsPath,omitempty"` - - // SMB-specific fields + // SMB-specific fields (also valid for type=smb routed through rclone-smb) SMBServer string `json:"smbServer,omitempty"` SMBShare string `json:"smbShare,omitempty"` SMBUsername string `json:"smbUsername,omitempty"` SMBPasswordObscured string `json:"smbPasswordObscured,omitempty"` SMBDomain string `json:"smbDomain,omitempty"` - // rclone-specific fields + // rclone-specific fields (escape hatch — references a pre-defined remote) RcloneRemote string `json:"rcloneRemote,omitempty"` RclonePath string `json:"rclonePath,omitempty"` + // VFS cache configuration — applied to the rclone mount call. All three + // fields are optional; blanks fall back to Default* constants above. + // CacheMaxSize uses rclone's SizeSuffix ("50G", "10M"), CacheMaxAge and + // DirCacheTime use rclone's Duration ("24h", "30m"). + CacheMaxSize string `json:"cacheMaxSize,omitempty"` + CacheMaxAge string `json:"cacheMaxAge,omitempty"` + DirCacheTime string `json:"dirCacheTime,omitempty"` + // Polling configuration PollingEnabled bool `json:"pollingEnabled"` PollingIntervalMs int `json:"pollingIntervalMs,omitempty"` // Default: 30000 @@ -56,6 +76,36 @@ type MountStatus struct { LastChecked int64 `json:"lastChecked"` PollingActive bool `json:"pollingActive,omitempty"` LastPolledScan int64 `json:"lastPolledScan,omitempty"` + // Progress tracking — surface the poller's adaptive scan state to the UI. + Scanning bool `json:"scanning,omitempty"` + CurrentScanStartedAt int64 `json:"currentScanStartedAt,omitempty"` // ms; non-zero only while Scanning + LastScanDurationMs int64 `json:"lastScanDurationMs,omitempty"` // duration of the most recently completed scan + NextScanAt int64 `json:"nextScanAt,omitempty"` // ms; planned start of the next adaptive scan + // Live IO stats — sampled by StatsPoller from /proc/fs/cifs/Stats, + // /proc/self/mountstats, or rclone's RC API depending on mount type. + IOStats *MountIOStats `json:"ioStats,omitempty"` +} + +// MountIOStats captures throughput + operation rate for a single mount. +// Cumulative counters (BytesRead, ReadOps) come straight from the source; +// rates (ReadBps, ReadIops) are computed by StatsPoller as the delta between +// the last two samples divided by the interval. +type MountIOStats struct { + Source string `json:"source"` // "rclone" — only source after the rclone-only consolidation + BytesRead int64 `json:"bytesRead"` + BytesWritten int64 `json:"bytesWritten,omitempty"` + ReadBps float64 `json:"readBps"` + WriteBps float64 `json:"writeBps,omitempty"` + ReadOps int64 `json:"readOps,omitempty"` + WriteOps int64 `json:"writeOps,omitempty"` + // ReadIops and WriteIops are op counters per second — number of read/write + // requests issued by the kernel against the share. Distinct from + // TransfersPerSec which is a file-level metric only relevant to rclone. + ReadIops float64 `json:"readIops,omitempty"` + WriteIops float64 `json:"writeIops,omitempty"` + TransfersPerSec float64 `json:"transfersPerSec,omitempty"` // rclone-only + LastSampleAt int64 `json:"lastSampleAt"` // ms + IntervalMs int64 `json:"intervalMs,omitempty"` // gap between the two samples used to compute rates } // MountsFile represents the mounts.json file structure @@ -72,14 +122,9 @@ type RcloneRemote struct { // CreateMountRequest is the request body for creating a mount type CreateMountRequest struct { - Name string `json:"name"` - Type MountType `json:"type"` - Enabled *bool `json:"enabled,omitempty"` // Pointer to detect if set - Options string `json:"options,omitempty"` - - // NFS - NFSServer string `json:"nfsServer,omitempty"` - NFSPath string `json:"nfsPath,omitempty"` + Name string `json:"name"` + Type MountType `json:"type"` + Enabled *bool `json:"enabled,omitempty"` // Pointer to detect if set // SMB SMBServer string `json:"smbServer,omitempty"` @@ -92,6 +137,11 @@ type CreateMountRequest struct { RcloneRemote string `json:"rcloneRemote,omitempty"` RclonePath string `json:"rclonePath,omitempty"` + // VFS cache (optional — falls back to Default* constants when unset) + CacheMaxSize string `json:"cacheMaxSize,omitempty"` + CacheMaxAge string `json:"cacheMaxAge,omitempty"` + DirCacheTime string `json:"dirCacheTime,omitempty"` + // Polling PollingEnabled *bool `json:"pollingEnabled,omitempty"` PollingIntervalMs *int `json:"pollingIntervalMs,omitempty"` diff --git a/internal/snapshot/cache.go b/internal/snapshot/cache.go new file mode 100644 index 0000000..0018b53 --- /dev/null +++ b/internal/snapshot/cache.go @@ -0,0 +1,143 @@ +package snapshot + +import ( + "archive/zip" + "errors" + "fmt" + "io" + "log" + "os" + "path/filepath" + "strings" +) + +// PluginOutputArchivePrefix is where plugin-output files live inside the ZIP. +// Trailing slash because ZIP entry names are forward-slash relative paths. +const PluginOutputArchivePrefix = "cache/plugin-output/" + +// addPluginOutput walks srcDir (typically $FILES_PATH/plugin) and copies every +// file under it into zw at PluginOutputArchivePrefix + relPath. +// +// Returns the number of files added. A missing srcDir is not an error: it +// just means no plugin has produced output yet, so the section stays empty. +func addPluginOutput(zw *zip.Writer, srcDir string) (int, error) { + info, err := os.Stat(srcDir) + if errors.Is(err, os.ErrNotExist) { + return 0, nil + } + if err != nil { + return 0, fmt.Errorf("stat %s: %w", srcDir, err) + } + if !info.IsDir() { + return 0, fmt.Errorf("%s is not a directory", srcDir) + } + + count := 0 + err = filepath.WalkDir(srcDir, func(path string, d os.DirEntry, walkErr error) error { + if walkErr != nil { + return walkErr + } + if d.IsDir() { + return nil + } + rel, err := filepath.Rel(srcDir, path) + if err != nil { + return err + } + entry := PluginOutputArchivePrefix + filepath.ToSlash(rel) + + w, err := zw.Create(entry) + if err != nil { + return err + } + f, err := os.Open(path) + if err != nil { + return err + } + defer f.Close() + if _, err := io.Copy(w, f); err != nil { + return err + } + count++ + return nil + }) + if err != nil { + return count, fmt.Errorf("walk %s: %w", srcDir, err) + } + return count, nil +} + +// extractPluginOutput pulls every entry under PluginOutputArchivePrefix from +// the ZIP and writes it under dstDir. Existing files are overwritten when +// overwrite=true; otherwise they're left alone (treats existing files as the +// "mine wins" choice — same toggle the metadata import uses). +// +// Path traversal is rejected: any entry whose relative path resolves outside +// dstDir is skipped with a warning. +func extractPluginOutput(zr *zip.Reader, dstDir string, overwrite bool) (written int, skipped int, err error) { + if dstDir == "" { + return 0, 0, fmt.Errorf("plugin-output destination dir is empty") + } + if err := os.MkdirAll(dstDir, 0o755); err != nil { + return 0, 0, fmt.Errorf("mkdir %s: %w", dstDir, err) + } + + cleanRoot, err := filepath.Abs(dstDir) + if err != nil { + return 0, 0, fmt.Errorf("abs %s: %w", dstDir, err) + } + + for _, f := range zr.File { + if !strings.HasPrefix(f.Name, PluginOutputArchivePrefix) { + continue + } + if strings.HasSuffix(f.Name, "/") { + continue + } + rel := strings.TrimPrefix(f.Name, PluginOutputArchivePrefix) + dst := filepath.Join(cleanRoot, filepath.FromSlash(rel)) + + // Reject anything that escapes the dst root. + if !strings.HasPrefix(dst, cleanRoot+string(os.PathSeparator)) && dst != cleanRoot { + log.Printf("[Snapshot] import: skipping path-escaping entry %q", f.Name) + skipped++ + continue + } + + if !overwrite { + if _, err := os.Stat(dst); err == nil { + skipped++ + continue + } + } + + if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil { + return written, skipped, fmt.Errorf("mkdir for %s: %w", dst, err) + } + + if err := writeZipFile(f, dst); err != nil { + return written, skipped, err + } + written++ + } + return written, skipped, nil +} + +func writeZipFile(f *zip.File, dst string) error { + rc, err := f.Open() + if err != nil { + return fmt.Errorf("open zip entry %s: %w", f.Name, err) + } + defer rc.Close() + + out, err := os.Create(dst) + if err != nil { + return fmt.Errorf("create %s: %w", dst, err) + } + defer out.Close() + + if _, err := io.Copy(out, rc); err != nil { + return fmt.Errorf("write %s: %w", dst, err) + } + return nil +} diff --git a/internal/snapshot/export.go b/internal/snapshot/export.go new file mode 100644 index 0000000..bdf2d01 --- /dev/null +++ b/internal/snapshot/export.go @@ -0,0 +1,132 @@ +package snapshot + +import ( + "archive/zip" + "encoding/json" + "fmt" + "io" + "log" + "os" + "time" + + "github.com/metazla/meta-core/internal/storage" +) + +// ExportOptions controls what an export includes. +type ExportOptions struct { + IncludePluginOutput bool + // PluginOutputDir is the absolute path on disk that holds plugin-produced + // artifacts (typically $FILES_PATH/plugin). Required when + // IncludePluginOutput is true. + PluginOutputDir string + // SourceHost is recorded in the manifest so imports can attribute + // where the snapshot came from (defaults to OS hostname). + SourceHost string +} + +// Exporter writes snapshot ZIPs from a live Redis store. +type Exporter struct { + storage *storage.Client +} + +func NewExporter(stor *storage.Client) *Exporter { + return &Exporter{storage: stor} +} + +// Export streams a ZIP archive to w. The archive layout is: +// +// manifest.json +// index.json +// metadata/{cid}.json (one per known CID) +// +// Optional sections (hash-index, plugin-output) are gated by ExportOptions +// but are not yet implemented — they live on filesystems meta-core does not +// own, and need cross-service collection. The flag is plumbed for forward +// compatibility but does nothing today. +func (e *Exporter) Export(w io.Writer, opts ExportOptions) (Manifest, error) { + if e.storage == nil || !e.storage.IsConnected() { + return Manifest{}, fmt.Errorf("storage not connected") + } + + hashIDs, err := e.storage.GetAllHashIDs() + if err != nil { + return Manifest{}, fmt.Errorf("list hashIDs: %w", err) + } + + zw := zip.NewWriter(w) + defer zw.Close() + + // 1. manifest.json — written last after we know FileCount, but we need + // the entry order to put it first in the archive for readers. Solve + // by buffering in memory: it's tiny. + host := opts.SourceHost + if host == "" { + if h, err := os.Hostname(); err == nil { + host = h + } + } + manifest := Manifest{ + SchemaVersion: ManifestSchemaVersion, + GeneratedAt: time.Now().UTC(), + SourceHost: host, + FileCount: len(hashIDs), + Includes: Includes{ + Metadata: true, + PluginOutput: opts.IncludePluginOutput, + }, + } + if err := writeJSON(zw, ManifestFile, manifest); err != nil { + return Manifest{}, fmt.Errorf("write manifest: %w", err) + } + + // 2. index.json — flat list of CIDs. Redundant with metadata/ but + // cheap and useful for readers that only need the keyset. + if err := writeJSON(zw, IndexFile, IndexFileBody{HashIDs: hashIDs}); err != nil { + return Manifest{}, fmt.Errorf("write index: %w", err) + } + + // 3. metadata/{cid}.json — one nested document per CID. + for _, hashID := range hashIDs { + flat, err := e.storage.GetMetadataFlat(hashID) + if err != nil { + log.Printf("[Snapshot] export: skipping %s: %v", hashID, err) + continue + } + if len(flat) == 0 { + continue + } + nested := Reconstruct(flat) + entry := MetadataDir + hashID + ".json" + if err := writeJSON(zw, entry, nested); err != nil { + return Manifest{}, fmt.Errorf("write %s: %w", entry, err) + } + } + + // 4. cache/plugin-output/* — optional. Walks the on-disk plugin dir and + // streams each file into the ZIP. Heavy (~120 MB on a populated dev + // stack), so it's gated by the ExportOptions toggle. + if opts.IncludePluginOutput { + if opts.PluginOutputDir == "" { + return Manifest{}, fmt.Errorf("PluginOutputDir is required when IncludePluginOutput is set") + } + n, err := addPluginOutput(zw, opts.PluginOutputDir) + if err != nil { + return Manifest{}, fmt.Errorf("add plugin-output: %w", err) + } + log.Printf("[Snapshot] exported %d plugin-output files from %s", n, opts.PluginOutputDir) + } + + log.Printf("[Snapshot] exported %d files (host=%s, pluginOutput=%v)", + len(hashIDs), host, opts.IncludePluginOutput) + return manifest, nil +} + +func writeJSON(zw *zip.Writer, name string, body interface{}) error { + w, err := zw.Create(name) + if err != nil { + return err + } + enc := json.NewEncoder(w) + enc.SetIndent("", " ") + return enc.Encode(body) +} diff --git a/internal/snapshot/flatten.go b/internal/snapshot/flatten.go new file mode 100644 index 0000000..603ed80 --- /dev/null +++ b/internal/snapshot/flatten.go @@ -0,0 +1,195 @@ +package snapshot + +import ( + "fmt" + "regexp" + "sort" + "strconv" + "strings" +) + +// Flat keys use "/" as the separator. Property paths inside Redis are +// stored as e.g. "fileinfo/streamdetails/video/0/codec". +const flatSep = "/" + +// numericKey matches a path segment that is a non-negative decimal integer. +// Used to detect array slots (".../video/0/codec"). +var numericKey = regexp.MustCompile(`^\d+$`) + +// Reconstruct rebuilds a nested JSON-shaped value from a flat property map. +// Keys are property paths (no leading slash, no file:{cid}/ prefix). +// +// Sibling keys whose path segments are all sequential integers starting at 0 +// are emitted as JSON arrays. Other groups become JSON objects. Leaf values +// are coerced from string to int/float/bool/null when they parse cleanly so +// the exported JSON is human-readable; on import we round-trip back to strings. +func Reconstruct(flat map[string]string) interface{} { + root := map[string]interface{}{} + for path, value := range flat { + parts := strings.Split(path, flatSep) + setPath(root, parts, parseValue(value)) + } + return convertNumericMapsToArrays(root) +} + +// Flatten is the inverse of Reconstruct: walks a nested value and produces a +// flat property map whose values are all strings (Redis-ready). +// +// Arrays become numeric path segments ("genre/0", "genre/1"). Nil/missing +// values are stored as "" to preserve the field's presence — this matches +// what flattenMetadata in the TS implementation does. +func Flatten(value interface{}) map[string]string { + out := map[string]string{} + flattenInto(value, "", out) + return out +} + +func setPath(root map[string]interface{}, parts []string, leaf interface{}) { + if len(parts) == 0 { + return + } + cur := root + for i, part := range parts { + if i == len(parts)-1 { + cur[part] = leaf + return + } + next, ok := cur[part].(map[string]interface{}) + if !ok { + next = map[string]interface{}{} + cur[part] = next + } + cur = next + } +} + +func convertNumericMapsToArrays(v interface{}) interface{} { + switch t := v.(type) { + case map[string]interface{}: + // Recurse first so child arrays are detected before parent. + for k, child := range t { + t[k] = convertNumericMapsToArrays(child) + } + if !looksLikeArray(t) { + return t + } + // Sort numerically and emit as []interface{}. + idx := make([]int, 0, len(t)) + for k := range t { + n, _ := strconv.Atoi(k) + idx = append(idx, n) + } + sort.Ints(idx) + arr := make([]interface{}, len(idx)) + for i, n := range idx { + arr[i] = t[strconv.Itoa(n)] + } + return arr + case []interface{}: + for i := range t { + t[i] = convertNumericMapsToArrays(t[i]) + } + return t + default: + return v + } +} + +// looksLikeArray returns true when every key in m is a non-negative integer +// and the keys form a contiguous range starting at 0. Empty maps return false +// — we can't tell whether the original was {} or [], so prefer {}. +func looksLikeArray(m map[string]interface{}) bool { + if len(m) == 0 { + return false + } + seen := make(map[int]bool, len(m)) + for k := range m { + if !numericKey.MatchString(k) { + return false + } + n, err := strconv.Atoi(k) + if err != nil || n < 0 { + return false + } + seen[n] = true + } + for i := 0; i < len(m); i++ { + if !seen[i] { + return false + } + } + return true +} + +func flattenInto(v interface{}, path string, out map[string]string) { + switch t := v.(type) { + case nil: + out[path] = "" + case map[string]interface{}: + if len(t) == 0 && path != "" { + // Empty objects don't have a Redis representation; skip. + return + } + for k, child := range t { + next := k + if path != "" { + next = path + flatSep + k + } + flattenInto(child, next, out) + } + case []interface{}: + for i, child := range t { + next := strconv.Itoa(i) + if path != "" { + next = path + flatSep + next + } + flattenInto(child, next, out) + } + case string: + out[path] = t + case bool: + if t { + out[path] = "true" + } else { + out[path] = "false" + } + case float64: + // JSON numbers always decode to float64; emit ints without decimals. + if t == float64(int64(t)) { + out[path] = strconv.FormatInt(int64(t), 10) + } else { + out[path] = strconv.FormatFloat(t, 'f', -1, 64) + } + default: + out[path] = fmt.Sprintf("%v", t) + } +} + +// parseValue mirrors the TS reconstruct logic: convert obvious primitives +// back to their typed form for nicer exported JSON, leave anything ambiguous +// as a string so we don't lose data (e.g. "5.1" channel layouts). +func parseValue(s string) interface{} { + if s == "" { + return nil + } + switch s { + case "true": + return true + case "false": + return false + case "null": + return nil + } + // Only treat as number when the string is a clean integer or decimal, + // no leading zeros / trailing junk. Matches the TS regex. + if isCleanNumber(s) { + if n, err := strconv.ParseFloat(s, 64); err == nil { + return n + } + } + return s +} + +var cleanNumber = regexp.MustCompile(`^-?\d+(\.\d+)?$`) + +func isCleanNumber(s string) bool { return cleanNumber.MatchString(s) } diff --git a/internal/snapshot/import.go b/internal/snapshot/import.go new file mode 100644 index 0000000..03d1329 --- /dev/null +++ b/internal/snapshot/import.go @@ -0,0 +1,258 @@ +package snapshot + +import ( + "archive/zip" + "encoding/json" + "fmt" + "io" + "log" + "strings" + + "github.com/metazla/meta-core/internal/storage" +) + +// Mode controls how an import reconciles existing CIDs. +type Mode string + +const ( + // ModeReplace deletes every existing property of an imported CID and + // writes the snapshot's values verbatim. Untouched CIDs are left alone. + ModeReplace Mode = "replace" + // ModeMerge keeps unmentioned existing properties and only touches the + // fields present in the snapshot. Conflict resolution within merge is + // controlled by Conflict. + ModeMerge Mode = "merge" +) + +// Conflict picks a winner for fields present on both sides during merge. +type Conflict string + +const ( + // ConflictMine keeps the existing (target) value when both sides have one. + ConflictMine Conflict = "mine" + // ConflictSource overwrites the existing value with the snapshot's value. + ConflictSource Conflict = "source" +) + +// ImportOptions controls how Import applies a snapshot. +type ImportOptions struct { + Mode Mode + Conflict Conflict + DryRun bool + // PluginOutputDir is the on-disk destination for cache/plugin-output/* + // entries. When empty, plugin-output is silently skipped on import. + PluginOutputDir string +} + +// FileResult summarizes what happened (or would happen, in dry-run) for one CID. +type FileResult struct { + HashID string `json:"hashId"` + Added int `json:"added"` // new property keys written + Updated int `json:"updated"` // existing property keys whose value changed + Unchanged int `json:"unchanged"` // existing property keys whose value matched + Kept int `json:"kept"` // existing property keys preserved by ConflictMine + Deleted int `json:"deleted"` // existing property keys removed (replace mode) + Error string `json:"error,omitempty"` +} + +// PluginOutputResult summarizes what happened to cache/plugin-output entries. +type PluginOutputResult struct { + Written int `json:"written"` + Skipped int `json:"skipped"` +} + +// ImportResult is the response body for the import endpoint. +type ImportResult struct { + Mode Mode `json:"mode"` + Conflict Conflict `json:"conflict,omitempty"` + DryRun bool `json:"dryRun"` + TotalFiles int `json:"totalFiles"` + FilesOK int `json:"filesOk"` + FilesFailed int `json:"filesFailed"` + Files []FileResult `json:"files,omitempty"` + PluginOutput *PluginOutputResult `json:"pluginOutput,omitempty"` +} + +// Importer applies snapshot ZIPs to a live Redis store. +type Importer struct { + storage *storage.Client +} + +func NewImporter(stor *storage.Client) *Importer { + return &Importer{storage: stor} +} + +// Import reads a ZIP from r (which must support io.ReaderAt; callers pass +// *bytes.Reader after buffering the upload) and applies it to Redis. +func (i *Importer) Import(r io.ReaderAt, size int64, opts ImportOptions) (*ImportResult, error) { + if i.storage == nil || !i.storage.IsConnected() { + return nil, fmt.Errorf("storage not connected") + } + if opts.Mode != ModeReplace && opts.Mode != ModeMerge { + return nil, fmt.Errorf("invalid mode %q (want replace|merge)", opts.Mode) + } + if opts.Mode == ModeMerge && opts.Conflict != ConflictMine && opts.Conflict != ConflictSource { + return nil, fmt.Errorf("invalid conflict %q (want mine|source)", opts.Conflict) + } + + zr, err := zip.NewReader(r, size) + if err != nil { + return nil, fmt.Errorf("open zip: %w", err) + } + + var manifest Manifest + if err := readJSON(zr, ManifestFile, &manifest); err != nil { + return nil, fmt.Errorf("read manifest: %w", err) + } + if manifest.SchemaVersion > ManifestSchemaVersion { + return nil, fmt.Errorf("snapshot schemaVersion %d exceeds supported %d", + manifest.SchemaVersion, ManifestSchemaVersion) + } + + result := &ImportResult{ + Mode: opts.Mode, + Conflict: opts.Conflict, + DryRun: opts.DryRun, + } + + for _, f := range zr.File { + if !strings.HasPrefix(f.Name, MetadataDir) || !strings.HasSuffix(f.Name, ".json") { + continue + } + hashID := strings.TrimSuffix(strings.TrimPrefix(f.Name, MetadataDir), ".json") + if hashID == "" { + continue + } + fr := i.applyOne(f, hashID, opts) + result.TotalFiles++ + if fr.Error != "" { + result.FilesFailed++ + } else { + result.FilesOK++ + } + result.Files = append(result.Files, fr) + } + + // Plugin-output extraction. Only applied on real imports (not dry-run). + // If the snapshot doesn't contain a cache/plugin-output/ section, this is + // a no-op. If the destination is unset, we skip silently — the manifest's + // pluginOutput flag tells the caller it was on the source side either way. + if !opts.DryRun && opts.PluginOutputDir != "" && manifest.Includes.PluginOutput { + // Existing files are kept under ConflictMine / replace-doesn't-apply, + // overwritten under ConflictSource. Replace mode behaves like + // "overwrite" for cache files (it's the destructive choice for + // metadata; mirror that intent here). + overwrite := opts.Mode == ModeReplace || opts.Conflict == ConflictSource + written, skipped, err := extractPluginOutput(zr, opts.PluginOutputDir, overwrite) + if err != nil { + return result, fmt.Errorf("extract plugin-output: %w", err) + } + result.PluginOutput = &PluginOutputResult{Written: written, Skipped: skipped} + log.Printf("[Snapshot] import: plugin-output written=%d skipped=%d", written, skipped) + } + + log.Printf("[Snapshot] import: mode=%s conflict=%s dryRun=%v files=%d ok=%d failed=%d", + opts.Mode, opts.Conflict, opts.DryRun, result.TotalFiles, result.FilesOK, result.FilesFailed) + return result, nil +} + +func (i *Importer) applyOne(f *zip.File, hashID string, opts ImportOptions) FileResult { + fr := FileResult{HashID: hashID} + + rc, err := f.Open() + if err != nil { + fr.Error = fmt.Sprintf("open entry: %v", err) + return fr + } + defer rc.Close() + + var nested interface{} + if err := json.NewDecoder(rc).Decode(&nested); err != nil { + fr.Error = fmt.Sprintf("decode json: %v", err) + return fr + } + incoming := Flatten(nested) + + existing, err := i.storage.GetMetadataFlat(hashID) + if err != nil { + fr.Error = fmt.Sprintf("read existing: %v", err) + return fr + } + if existing == nil { + existing = map[string]string{} + } + + switch opts.Mode { + case ModeReplace: + // Diff for reporting. + toWrite := map[string]string{} + for k, v := range incoming { + toWrite[k] = v + if cur, ok := existing[k]; ok { + if cur == v { + fr.Unchanged++ + } else { + fr.Updated++ + } + } else { + fr.Added++ + } + } + for k := range existing { + if _, ok := incoming[k]; !ok { + fr.Deleted++ + } + } + if !opts.DryRun { + if _, err := i.storage.DeleteMetadata(hashID); err != nil { + fr.Error = fmt.Sprintf("delete existing: %v", err) + return fr + } + if err := i.storage.SetMetadataFlat(hashID, toWrite); err != nil { + fr.Error = fmt.Sprintf("write: %v", err) + return fr + } + } + + case ModeMerge: + toWrite := map[string]string{} + for k, v := range incoming { + cur, present := existing[k] + switch { + case !present: + toWrite[k] = v + fr.Added++ + case cur == v: + fr.Unchanged++ + case opts.Conflict == ConflictSource: + toWrite[k] = v + fr.Updated++ + case opts.Conflict == ConflictMine: + fr.Kept++ + } + } + if !opts.DryRun && len(toWrite) > 0 { + if _, err := i.storage.MergeMetadataFlat(hashID, toWrite); err != nil { + fr.Error = fmt.Sprintf("merge: %v", err) + return fr + } + } + } + + return fr +} + +func readJSON(zr *zip.Reader, name string, out interface{}) error { + for _, f := range zr.File { + if f.Name != name { + continue + } + rc, err := f.Open() + if err != nil { + return err + } + defer rc.Close() + return json.NewDecoder(rc).Decode(out) + } + return fmt.Errorf("entry %q not found", name) +} diff --git a/internal/snapshot/manifest.go b/internal/snapshot/manifest.go new file mode 100644 index 0000000..9877316 --- /dev/null +++ b/internal/snapshot/manifest.go @@ -0,0 +1,42 @@ +// Package snapshot implements export/import/wipe of MetaMesh metadata. +// +// The on-disk format is a ZIP archive containing one nested-JSON file per +// known CID (cid_midhash256). Schema is defined by ManifestSchemaVersion; +// readers must reject archives whose schemaVersion exceeds what they know. +package snapshot + +import "time" + +// ManifestSchemaVersion is the on-disk format version. Bump when the layout +// of metadata/{cid}.json or the manifest changes in a non-additive way. +const ManifestSchemaVersion = 1 + +// MetadataDir is the path inside the ZIP that holds per-CID JSON files. +const MetadataDir = "metadata/" + +// ManifestFile is the path inside the ZIP that holds the manifest JSON. +const ManifestFile = "manifest.json" + +// IndexFile is the path inside the ZIP that holds the file index (CID list). +const IndexFile = "index.json" + +// Manifest is the metadata header stored at the root of every snapshot ZIP. +type Manifest struct { + SchemaVersion int `json:"schemaVersion"` + GeneratedAt time.Time `json:"generatedAt"` + SourceHost string `json:"sourceHost,omitempty"` + MetaCoreVer string `json:"metaCoreVersion,omitempty"` + FileCount int `json:"fileCount"` + Includes Includes `json:"includes"` +} + +// Includes records which optional sections are present in a snapshot. +type Includes struct { + Metadata bool `json:"metadata"` + PluginOutput bool `json:"pluginOutput,omitempty"` +} + +// IndexFileBody is the structure of index.json inside the ZIP. +type IndexFileBody struct { + HashIDs []string `json:"hashIds"` +} diff --git a/internal/snapshot/wipe.go b/internal/snapshot/wipe.go new file mode 100644 index 0000000..21d6172 --- /dev/null +++ b/internal/snapshot/wipe.go @@ -0,0 +1,42 @@ +package snapshot + +import ( + "fmt" + + "github.com/metazla/meta-core/internal/storage" +) + +// WipeScope picks which classes of data to clear. +type WipeScope struct { + Metadata bool // file:* keys + file:__index__ +} + +// WipeResult reports what was cleared. +type WipeResult struct { + MetadataDeleted int64 `json:"metadataDeleted"` +} + +// Wiper clears metadata from Redis. +type Wiper struct { + storage *storage.Client +} + +func NewWiper(stor *storage.Client) *Wiper { + return &Wiper{storage: stor} +} + +// Wipe applies the requested scope. Returns the count of files removed. +func (w *Wiper) Wipe(scope WipeScope) (*WipeResult, error) { + if w.storage == nil || !w.storage.IsConnected() { + return nil, fmt.Errorf("storage not connected") + } + res := &WipeResult{} + if scope.Metadata { + n, err := w.storage.ClearAllMetadata() + if err != nil { + return nil, fmt.Errorf("clear metadata: %w", err) + } + res.MetadataDeleted = n + } + return res, nil +} diff --git a/internal/storage/client.go b/internal/storage/client.go index 90d30b8..6291089 100644 --- a/internal/storage/client.go +++ b/internal/storage/client.go @@ -537,6 +537,93 @@ func (c *Client) RemoveFromSet(hashID, property, value string) (bool, error) { return true, c.client.Set(ctx, key, newValue, 0).Err() } +// FileTuple holds the (path, size, mtime) record for one file. Used by the +// watcher hydrator at startup to avoid re-mid-hashing files whose tuple +// matches what's already in Redis. +type FileTuple struct { + HashID string + FilePath string + Size int64 + MtimeNano int64 +} + +// GetTuplesForAllFiles returns one FileTuple per indexed CID, fetched in +// batched MGETs (no SCAN per CID). Skips files missing any of the three +// fields. Sub-second on local Redis for ~10K files. +func (c *Client) GetTuplesForAllFiles() ([]FileTuple, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return nil, fmt.Errorf("not connected") + } + + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + + hashIDs, err := c.client.SMembers(ctx, c.buildIndexKey()).Result() + if err != nil { + return nil, fmt.Errorf("smembers index: %w", err) + } + if len(hashIDs) == 0 { + return nil, nil + } + + // Build the key list: 3 keys per CID, in a fixed order so we can decode + // the MGET response by position. fields[i] for hashIDs[j] sits at + // keys[j*3+i]. + const fieldsPerFile = 3 + fields := [fieldsPerFile]string{"filePath", "sizeByte", "mtimeNano"} + keys := make([]string, 0, len(hashIDs)*fieldsPerFile) + for _, h := range hashIDs { + prefix := c.buildKeyPrefix(h) + for _, f := range fields { + keys = append(keys, prefix+f) + } + } + + // Chunked MGET — Redis caps args per command; 1000 is a safe batch. + const batch = 1000 + values := make([]interface{}, 0, len(keys)) + for start := 0; start < len(keys); start += batch { + end := start + batch + if end > len(keys) { + end = len(keys) + } + chunk, err := c.client.MGet(ctx, keys[start:end]...).Result() + if err != nil { + return nil, fmt.Errorf("mget chunk [%d:%d]: %w", start, end, err) + } + values = append(values, chunk...) + } + + out := make([]FileTuple, 0, len(hashIDs)) + for i, h := range hashIDs { + base := i * fieldsPerFile + filePath, _ := values[base].(string) + sizeStr, _ := values[base+1].(string) + mtimeStr, _ := values[base+2].(string) + if filePath == "" || sizeStr == "" || mtimeStr == "" { + continue + } + size, err := strconv.ParseInt(sizeStr, 10, 64) + if err != nil { + continue + } + mtime, err := strconv.ParseInt(mtimeStr, 10, 64) + if err != nil { + continue + } + out = append(out, FileTuple{ + HashID: h, + FilePath: filePath, + Size: size, + MtimeNano: mtime, + }) + } + return out, nil +} + // GetMemoryInfo returns Redis memory usage information func (c *Client) GetMemoryInfo() (string, error) { c.mu.RLock() @@ -1059,6 +1146,51 @@ func (c *Client) ClearStream(stream string) error { return nil } +// StreamEntry represents one entry from a Redis stream. +type StreamEntry struct { + ID string + Values map[string]interface{} +} + +// XRevRange reads stream entries from newest to oldest. +// start defaults to "+" (newest) and stop defaults to "-" (oldest) when empty. +// If count > 0, at most that many entries are returned. +func (c *Client) XRevRange(stream, start, stop string, count int64) ([]StreamEntry, error) { + c.mu.RLock() + defer c.mu.RUnlock() + + if c.client == nil { + return nil, fmt.Errorf("not connected") + } + + if start == "" { + start = "+" + } + if stop == "" { + stop = "-" + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + var msgs []redis.XMessage + var err error + if count > 0 { + msgs, err = c.client.XRevRangeN(ctx, stream, start, stop, count).Result() + } else { + msgs, err = c.client.XRevRange(ctx, stream, start, stop).Result() + } + if err != nil { + return nil, fmt.Errorf("xrevrange failed: %w", err) + } + + entries := make([]StreamEntry, 0, len(msgs)) + for _, m := range msgs { + entries = append(entries, StreamEntry{ID: m.ID, Values: m.Values}) + } + return entries, nil +} + // GetStreamLength returns the length of a Redis stream func (c *Client) GetStreamLength(stream string) (int64, error) { c.mu.RLock() diff --git a/internal/watcher/dispatcher.go b/internal/watcher/dispatcher.go index 46ab142..429a629 100644 --- a/internal/watcher/dispatcher.go +++ b/internal/watcher/dispatcher.go @@ -3,6 +3,8 @@ package watcher import ( "fmt" "log" + "strconv" + "strings" "sync" "github.com/metazla/meta-core/internal/storage" @@ -108,6 +110,85 @@ func (d *Dispatcher) EmitReset(watcherID string) error { return nil } +// GetRecentEvents reads recent events from the Redis stream, newest first. +// If sinceMS > 0, only events strictly newer than that timestamp are returned. +// limit caps the number of returned events (defaults to 100 when <= 0). +func (d *Dispatcher) GetRecentEvents(sinceMS int64, limit int) ([]FileEvent, error) { + d.mu.RLock() + client := d.storageClient + d.mu.RUnlock() + + if client == nil || !client.IsConnected() { + return []FileEvent{}, nil + } + + if limit <= 0 { + limit = 100 + } + + start := "+" + stop := "-" + if sinceMS > 0 { + // Exclusive lower bound: skip entries at or before sinceMS. + stop = fmt.Sprintf("(%d-0", sinceMS) + } + + entries, err := client.XRevRange(EventsStream, start, stop, int64(limit)) + if err != nil { + return nil, err + } + + events := make([]FileEvent, 0, len(entries)) + for _, entry := range entries { + events = append(events, parseStreamEntry(entry)) + } + return events, nil +} + +// parseStreamEntry converts a Redis stream entry back into a FileEvent. +// All fields are stored as strings by XADD, so numerics need parsing. +func parseStreamEntry(entry storage.StreamEntry) FileEvent { + event := FileEvent{} + + if v, ok := entry.Values["type"].(string); ok { + event.Type = FileEventType(v) + } + if v, ok := entry.Values["path"].(string); ok { + event.Path = v + } + if v, ok := entry.Values["timestamp"].(string); ok { + if n, err := strconv.ParseInt(v, 10, 64); err == nil { + event.Timestamp = n + } + } + if v, ok := entry.Values["size"].(string); ok { + if n, err := strconv.ParseInt(v, 10, 64); err == nil { + event.Size = n + } + } + if v, ok := entry.Values["midhash256"].(string); ok { + event.MidHash256 = v + } + if v, ok := entry.Values["oldPath"].(string); ok { + event.OldPath = v + } + if v, ok := entry.Values["watcherId"].(string); ok { + event.WatcherID = v + } + + // Fall back to the stream ID's millisecond prefix if the timestamp + // field is missing or unparseable — every Redis stream ID starts with ms. + if event.Timestamp == 0 && entry.ID != "" { + if dash := strings.IndexByte(entry.ID, '-'); dash > 0 { + if n, err := strconv.ParseInt(entry.ID[:dash], 10, 64); err == nil { + event.Timestamp = n + } + } + } + + return event +} + // GetStreamLength returns the current stream length func (d *Dispatcher) GetStreamLength() (int64, error) { d.mu.RLock() diff --git a/internal/watcher/handlers.go b/internal/watcher/handlers.go index ef2ce11..5e71218 100644 --- a/internal/watcher/handlers.go +++ b/internal/watcher/handlers.go @@ -2,6 +2,7 @@ package watcher import ( "encoding/json" + "log" "net/http" "strconv" @@ -52,7 +53,14 @@ func (h *Handlers) handlePoll(w http.ResponseWriter, r *http.Request) { } } - events := h.watcher.GetRecentEvents(sinceMS, limit) + // Read events from the Redis `file:events` stream rather than the + // in-memory buffer. The buffer is fed by the unused debouncer path, + // so it is always empty in practice — the stream is the source of truth. + events, err := h.dispatcher.GetRecentEvents(sinceMS, limit) + if err != nil { + log.Printf("[watcher.handlePoll] failed to read stream: %v", err) + events = []FileEvent{} + } writeJSON(w, http.StatusOK, EventsListResponse{ Events: events, diff --git a/internal/watcher/midhash.go b/internal/watcher/midhash.go index 8154055..f60912b 100644 --- a/internal/watcher/midhash.go +++ b/internal/watcher/midhash.go @@ -26,15 +26,13 @@ const ( // The size prefix (8-byte big-endian) ensures that files with identical // middle content but different sizes produce different hashes. // +// fileSize must be passed in by the caller (typically from a FileInfo it +// already has from filepath.Walk or a previous os.Stat). This function +// deliberately does NOT call os.Stat itself — re-statting every file during +// a scan doubles the syscall count, which matters on slow / network drives. +// // Returns: CID v1 string in base32lower format (prefixed with "b") -func ComputeMidHash256(filePath string) (string, error) { - // Get file info - fileInfo, err := os.Stat(filePath) - if err != nil { - return "", err - } - fileSize := fileInfo.Size() - +func ComputeMidHash256(filePath string, fileSize int64) (string, error) { // Open file file, err := os.Open(filePath) if err != nil { diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index f13ce64..6ba4d36 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -4,10 +4,12 @@ import ( "log" "os" "path/filepath" + "strconv" "strings" "sync" "github.com/metazla/meta-core/internal/config" + "github.com/metazla/meta-core/internal/storage" ) @@ -18,11 +20,39 @@ type Watcher struct { dispatcher *Dispatcher filesPath string stateRegistry *StateRegistry + storage *storage.Client mu sync.RWMutex eventBuffer []FileEvent } +// SetStorageClient lets the watcher persist the (path, size, mtime, midhash) +// tuple to Redis whenever it computes a midhash. Without this, only the +// in-memory state registry knows about the file — Redis would only learn +// about it via downstream consumers of the event stream. +func (w *Watcher) SetStorageClient(s *storage.Client) { + w.storage = s +} + +// writeFileTuple persists path/size/mtime/midhash for a freshly-hashed file. +// Caller already has all four values from its existing os.Stat / FileInfo — +// no extra IO is performed here beyond the Redis write itself. +// +// This is the inverse-index that lets a fresh meta-core boot skip +// re-mid-hashing files whose (path, size, mtime) match what's in Redis. +func (w *Watcher) writeFileTuple(absPath string, size, mtimeNano int64, midhash string) { + if w.storage == nil || midhash == "" { + return + } + if _, err := w.storage.MergeMetadataFlat(midhash, map[string]string{ + "filePath": absPath, + "sizeByte": strconv.FormatInt(size, 10), + "mtimeNano": strconv.FormatInt(mtimeNano, 10), + }); err != nil { + log.Printf("[Watcher] failed to write tuple for %s: %v", midhash, err) + } +} + // NewWatcher creates a new file watcher/scanner func NewWatcher(cfg *config.Config, dispatcher *Dispatcher) (*Watcher, error) { debouncer := NewDebouncer(0) // No debouncing for polling-based scanning @@ -44,7 +74,7 @@ func (w *Watcher) handleDebouncedEvent(event FileEvent) { // Compute midhash256 for add/change events if event.Type == EventTypeAdd || event.Type == EventTypeChange { fullPath := filepath.Join(w.filesPath, event.Path) - if hash, err := ComputeMidHash256(fullPath); err == nil { + if hash, err := ComputeMidHash256(fullPath, event.Size); err == nil { event.MidHash256 = hash } } @@ -90,16 +120,20 @@ func (w *Watcher) ScanDirectory(root string) int { relPath = filepath.ToSlash(relPath) // Emit add event + curSize := info.Size() + curMtime := info.ModTime().UnixNano() event := FileEvent{ Type: EventTypeAdd, Path: relPath, - Size: info.Size(), + Size: curSize, Timestamp: NowMS(), } - // Compute midhash256 - if hash, err := ComputeMidHash256(path); err == nil { + // Compute midhash256 — uses the size we already have from Walk's + // FileInfo, so no extra os.Stat is needed. + if hash, err := ComputeMidHash256(path, curSize); err == nil { event.MidHash256 = hash + w.writeFileTuple(path, curSize, curMtime, hash) } // Dispatch directly @@ -160,9 +194,11 @@ func (w *Watcher) ScanDirectoryDiff(root string, watcherID string) DiffScanResul Timestamp: NowMS(), } - // Compute midhash256 - if hash, err := ComputeMidHash256(path); err == nil { + // Compute midhash256 — size is already known from the walk's + // FileInfo, no extra stat needed. + if hash, err := ComputeMidHash256(path, currentSize); err == nil { event.MidHash256 = hash + w.writeFileTuple(path, currentSize, currentMtime, hash) } // Dispatch event @@ -185,9 +221,10 @@ func (w *Watcher) ScanDirectoryDiff(root string, watcherID string) DiffScanResul Timestamp: NowMS(), } - // Compute midhash256 - if hash, err := ComputeMidHash256(path); err == nil { + // Compute midhash256 — reuse the size we already have. + if hash, err := ComputeMidHash256(path, currentSize); err == nil { event.MidHash256 = hash + w.writeFileTuple(path, currentSize, currentMtime, hash) } // Dispatch event @@ -227,6 +264,75 @@ func (w *Watcher) ScanDirectoryDiff(root string, watcherID string) DiffScanResul return result } +// WatcherRoot pairs a watcher's id with the absolute path it watches. The +// hydrator uses Root to decide which Redis-tracked file belongs to which +// in-memory WatcherState. +type WatcherRoot struct { + ID string + Root string +} + +// HydrateStateFromStorage repopulates the per-watcher in-memory state from +// the (filePath, sizeByte, mtimeNano) tuples in Redis. After this runs, the +// next scan tick will see existing files as "known" and skip +// ComputeMidHash256 for any whose size+mtime match — turning a cold-start +// rescan from minutes to seconds on a populated library. +// +// Files whose filePath doesn't fall under any watcher's root are silently +// skipped; files missing any tuple field are skipped too (a partial record +// can't seed valid state). +// +// Returns counts so the caller can log / surface them. +func (w *Watcher) HydrateStateFromStorage(roots []WatcherRoot) (loaded, skipped int, err error) { + if w.storage == nil || !w.storage.IsConnected() { + return 0, 0, nil + } + + tuples, err := w.storage.GetTuplesForAllFiles() + if err != nil { + return 0, 0, err + } + + // Sort roots by length descending so longest-prefix wins when paths + // nest (e.g. /files/watch/foo and /files/watch). + sortedRoots := make([]WatcherRoot, len(roots)) + copy(sortedRoots, roots) + for i := 1; i < len(sortedRoots); i++ { + for j := i; j > 0 && len(sortedRoots[j].Root) > len(sortedRoots[j-1].Root); j-- { + sortedRoots[j], sortedRoots[j-1] = sortedRoots[j-1], sortedRoots[j] + } + } + + // State keys are relative to w.filesPath (matching ScanDirectoryDiff's + // Rel(w.filesPath, path) convention). The watcher root only decides + // *which* per-watcher state map this file lives in. + for _, t := range tuples { + matched := false + for _, r := range sortedRoots { + if t.FilePath == r.Root || strings.HasPrefix(t.FilePath, r.Root+"/") { + rel, err := filepath.Rel(w.filesPath, t.FilePath) + if err != nil { + rel = t.FilePath + } + rel = filepath.ToSlash(rel) + ws := w.stateRegistry.GetOrCreate(r.ID) + ws.Set(rel, &FileState{ + Size: t.Size, + MtimeNano: t.MtimeNano, + MidHash: t.HashID, + }) + loaded++ + matched = true + break + } + } + if !matched { + skipped++ + } + } + return loaded, skipped, nil +} + // ClearState clears the tracked state for a watcher func (w *Watcher) ClearState(watcherID string) { w.stateRegistry.Clear(watcherID) diff --git a/meta-core.dev b/meta-core.dev new file mode 100644 index 0000000..43fed7b Binary files /dev/null and b/meta-core.dev differ