diff options
| -rw-r--r-- | cmd_linux.go | 255 | ||||
| -rw-r--r-- | cmd_windows.go | 165 | ||||
| -rw-r--r-- | types.go | 54 |
3 files changed, 391 insertions, 83 deletions
diff --git a/cmd_linux.go b/cmd_linux.go index 9e7e237..5cb48a4 100644 --- a/cmd_linux.go +++ b/cmd_linux.go @@ -5,7 +5,6 @@ package main import ( "bufio" "context" - "encoding/json" "flag" "net" "os" @@ -16,12 +15,12 @@ import ( ) var ( - hostDir = flag.String("host", "/host", "Host source directory (read-only)") - dataDir = flag.String("data", "/data", "Data destination directory") - verbose = flag.Bool("verbose", false, "Enable verbose logging (shorthand for -log-level debug)") - logLevel = flag.String("log-level", "info", "Log level: debug, info, warn, error, none") - dryRun = flag.Bool("dry-run", false, "Preview what would be synced without copying") - socketPath = flag.String("socket", "", "Unix socket to receive events from (Windows watcher)") + hostDir = flag.String("host", "/host", "Host source directory (read-only)") + dataDir = flag.String("data", "/data", "Data destination directory") + verbose = flag.Bool("verbose", false, "Enable verbose logging (shorthand for -log-level debug)") + logLevel = flag.String("log-level", "info", "Log level: debug, info, warn, error, none") + dryRun = flag.Bool("dry-run", false, "Preview what would be synced without copying") + socketPath = flag.String("socket", "", "Unix socket to receive events from (Windows watcher)") ) func run() { @@ -38,94 +37,240 @@ func run() { } SetLogLevel(level) - stack := NewGitignoreStack(*hostDir) - - manifest, err := initialSync(*hostDir, *dataDir, stack, *dryRun) - if err != nil { - logErrorf("initial sync failed: %v", err) - os.Exit(1) - } + ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer cancel() - if *dryRun { - logInfo("dry-run complete, exiting") - return - } + stack := NewGitignoreStack(*hostDir) if *socketPath == "" { - logInfo("no -socket specified, running in standalone mode (no file watching)") - ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - defer cancel() + logInfo("no -socket specified, running standalone (full scan)") + manifest, err := initialSync(*hostDir, *dataDir, stack, *dryRun) + if err != nil { + logErrorf("initial sync failed: %v", err) + os.Exit(1) + } + if *dryRun { + logInfo("dry-run complete, exiting") + return + } + logInfo("watcher: ready (no file watching in standalone mode)") + _ = manifest <-ctx.Done() return } logInfo("watcher: connecting to %s ...", *socketPath) - ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) - defer cancel() - if err := listenSocket(ctx, *socketPath, manifest, stack); err != nil { - logErrorf("socket listener failed: %v", err) - os.Exit(1) - } -} - -func listenSocket(ctx context.Context, socketPath string, manifest *Manifest, stack *GitignoreStack) error { var conn net.Conn - var err error - for { if ctx.Err() != nil { - return ctx.Err() + logErrorf("cancelled while connecting") + os.Exit(1) } - conn, err = net.Dial("unix", socketPath) + conn, err = net.Dial("unix", *socketPath) if err == nil { break } logDebug("connect failed: %v, retrying...", err) select { case <-ctx.Done(): - return ctx.Err() + logErrorf("cancelled while connecting") + os.Exit(1) case <-time.After(1 * time.Second): } } defer conn.Close() - logInfo("watcher: connected, monitoring for changes ...") + logInfo("watcher: connected, receiving manifest ...") scanner := bufio.NewScanner(conn) + remoteManifest := &Manifest{Files: make(map[string]FileMeta)} + for scanner.Scan() { - var evt FileEvent - if err := json.Unmarshal(scanner.Bytes(), &evt); err != nil { - logDebug("bad event: %v", err) + msg, err := DecodeMessage(scanner.Bytes()) + if err != nil { + logDebug("bad message: %v", err) + continue + } + + switch msg.Type { + case "manifest": + remoteManifest = ManifestFromJSON(msg.Files) + logInfo("manifest: received %d files from host", len(remoteManifest.Files)) + + case "manifest_done": + logInfo("manifest: comparing with local state ...") + syncFromManifest(remoteManifest, stack) + logInfo("sync: done") + logInfo("watcher: monitoring for changes ...") + + case "event_create": + logDebug("event: create %s", msg.Path) + if !isGitDir(msg.Path) && !stack.IsIgnored(msg.Path) { + syncCreate( + filepath.Join(*hostDir, msg.Path), + filepath.Join(*dataDir, msg.Path), + msg.Path, remoteManifest, stack, *dryRun, nil, + ) + } + + case "event_write": + logDebug("event: write %s", msg.Path) + if !isGitDir(msg.Path) && !stack.IsIgnored(msg.Path) { + syncWrite( + filepath.Join(*hostDir, msg.Path), + filepath.Join(*dataDir, msg.Path), + msg.Path, remoteManifest, *dryRun, + ) + } + + case "event_remove": + logDebug("event: remove %s", msg.Path) + if !isGitDir(msg.Path) && !stack.IsIgnored(msg.Path) { + syncRemove( + filepath.Join(*dataDir, msg.Path), + msg.Path, remoteManifest, *dryRun, + ) + } + } + } + + if err := scanner.Err(); err != nil { + logErrorf("socket read: %v", err) + } +} + +func syncFromManifest(remote *Manifest, stack *GitignoreStack) { + stats := &syncStats{} + start := time.Now() + + // Scan local /data to get current state (fast, local volume) + localManifest := scanLocalDirectory(*dataDir) + + for relPath, remoteMeta := range remote.Files { + if isGitDir(relPath) { + continue + } + if stack.IsIgnored(relPath) { + stats.skipped++ continue } - logDebug("event: %s %s", evt.Op, evt.Path) + destPath := filepath.Join(*dataDir, relPath) + srcPath := filepath.Join(*hostDir, relPath) - if isGitDir(evt.Path) { + // Check if file already exists locally with same metadata + if localMeta, exists := localManifest.Files[relPath]; exists { + if localMeta.Size == remoteMeta.Size && + localMeta.Mode == remoteMeta.Mode && + localMeta.ModTime.Equal(remoteMeta.ModTime) && + localMeta.Symlink == remoteMeta.Symlink { + continue + } + } + + // Need to sync + info, err := os.Lstat(srcPath) + if err != nil { + if os.IsNotExist(err) { + continue + } + stats.errors++ + logErrorf("stat %s: %v", relPath, err) continue } - if stack.IsIgnored(evt.Path) { + if info.IsDir() { + stats.dirs++ + if *dryRun { + logInfo("[dry-run] would create dir: %s", relPath) + continue + } + if err := os.MkdirAll(destPath, info.Mode()); err != nil { + stats.errors++ + logErrorf("mkdir %s: %v", relPath, err) + } continue } - destPath := filepath.Join(*dataDir, evt.Path) - srcPath := filepath.Join(*hostDir, evt.Path) + stats.files++ - switch evt.Op { - case "create": - syncCreate(srcPath, destPath, evt.Path, manifest, stack, *dryRun, nil) - case "write": - syncWrite(srcPath, destPath, evt.Path, manifest, *dryRun) - case "remove": - syncRemove(destPath, evt.Path, manifest, *dryRun) + if *dryRun { + logInfo("[dry-run] would sync: %s", relPath) + continue } - } - if err := scanner.Err(); err != nil { - return err + os.MkdirAll(filepath.Dir(destPath), 0755) + os.Remove(destPath) + + linkTarget := "" + if info.Mode()&os.ModeSymlink != 0 { + linkTarget, err = os.Readlink(srcPath) + if err != nil { + stats.errors++ + logErrorf("readlink %s: %v", relPath, err) + continue + } + if err := os.Symlink(linkTarget, destPath); err != nil { + stats.errors++ + logErrorf("symlink %s: %v", relPath, err) + } + continue + } + + if err := copyFileWithRetry(srcPath, destPath, info.Mode()); err != nil { + stats.errors++ + logErrorf("copy %s: %v", relPath, err) + } + + if stats.files%1000 == 0 && stats.files > 0 { + logInfo(" synced %d files ...", stats.files) + } } - return nil + elapsed := time.Since(start) + logInfo("sync: done in %s — %d files, %d dirs, %d skipped, %d errors", + elapsed.Round(time.Millisecond), stats.files, stats.dirs, stats.skipped, stats.errors) +} + +func scanLocalDirectory(root string) *Manifest { + manifest := &Manifest{Files: make(map[string]FileMeta)} + + filepath.Walk(root, func(path string, info os.FileInfo, err error) error { + if err != nil { + return nil + } + + relPath, err := filepath.Rel(root, path) + if err != nil { + return nil + } + + if relPath == "." { + return nil + } + + if isGitDir(relPath) { + if info.IsDir() { + return filepath.SkipDir + } + return nil + } + + linkTarget := "" + if info.Mode()&os.ModeSymlink != 0 { + linkTarget, _ = os.Readlink(path) + } + + manifest.Files[relPath] = FileMeta{ + Size: info.Size(), + Mode: info.Mode(), + ModTime: info.ModTime(), + Symlink: linkTarget, + } + + return nil + }) + + return manifest } diff --git a/cmd_windows.go b/cmd_windows.go index b68851a..61c7f1d 100644 --- a/cmd_windows.go +++ b/cmd_windows.go @@ -3,8 +3,8 @@ package main import ( + "bufio" "context" - "encoding/json" "flag" "fmt" "net" @@ -38,10 +38,12 @@ func run() { } SetLogLevel(level) - logInfo("watcher: starting on %s", *sourceDir) - logInfo("watcher: serving on %s", *socketPath) + logInfo("watcher: scanning %s ...", *sourceDir) stack := NewGitignoreStack(*sourceDir) + manifest := scanDirectory(*sourceDir, stack) + + logInfo("watcher: found %d files, serving on %s", len(manifest.Files), *socketPath) watcher, err := fsnotify.NewWatcher() if err != nil { @@ -86,40 +88,25 @@ func run() { logDebug("client connected: %s", conn.RemoteAddr()) go func(c net.Conn) { - defer func() { - clientsMu.Lock() - delete(clients, c) - clientsMu.Unlock() - c.Close() - }() - buf := make([]byte, 1) - for { - if _, err := c.Read(buf); err != nil { - return - } + sendManifest(c, manifest) + + scanner := bufio.NewScanner(c) + for scanner.Scan() { } + + clientsMu.Lock() + delete(clients, c) + clientsMu.Unlock() + c.Close() }(conn) } }() broadcast := func(evt FileEvent) { - data, err := json.Marshal(evt) - if err != nil { - return - } - data = append(data, '\n') - clientsMu.Lock() defer clientsMu.Unlock() for c := range clients { - if _, err := c.Write(data); err != nil { - go func() { - clientsMu.Lock() - delete(clients, c) - clientsMu.Unlock() - c.Close() - }() - } + sendEvent(c, evt) } } @@ -205,6 +192,39 @@ func run() { return } + // Update manifest + srcPath := filepath.Join(*sourceDir, path) + switch evt.Op { + case "create": + if info, err := os.Lstat(srcPath); err == nil { + linkTarget := "" + if info.Mode()&os.ModeSymlink != 0 { + linkTarget, _ = os.Readlink(srcPath) + } + manifest.Files[path] = FileMeta{ + Size: info.Size(), + Mode: info.Mode(), + ModTime: info.ModTime(), + Symlink: linkTarget, + } + } + case "write": + if info, err := os.Lstat(srcPath); err == nil { + linkTarget := "" + if info.Mode()&os.ModeSymlink != 0 { + linkTarget, _ = os.Readlink(srcPath) + } + manifest.Files[path] = FileMeta{ + Size: info.Size(), + Mode: info.Mode(), + ModTime: info.ModTime(), + Symlink: linkTarget, + } + } + case "remove": + delete(manifest.Files, path) + } + broadcast(evt) }) mu.Unlock() @@ -218,6 +238,95 @@ func run() { } } +func scanDirectory(root string, stack *GitignoreStack) *Manifest { + manifest := &Manifest{Files: make(map[string]FileMeta)} + + filepath.Walk(root, func(path string, info os.FileInfo, err error) error { + if err != nil { + return nil + } + + relPath, err := filepath.Rel(root, path) + if err != nil { + return nil + } + + if relPath == "." { + return nil + } + + if isGitDir(relPath) { + if info.IsDir() { + return filepath.SkipDir + } + return nil + } + + if info.IsDir() && hasGitignore(path) { + stack.Push(path) + } + + if stack.IsIgnored(relPath) { + return nil + } + + linkTarget := "" + if info.Mode()&os.ModeSymlink != 0 { + linkTarget, _ = os.Readlink(path) + } + + manifest.Files[relPath] = FileMeta{ + Size: info.Size(), + Mode: info.Mode(), + ModTime: info.ModTime(), + Symlink: linkTarget, + } + + return nil + }) + + return manifest +} + +func sendManifest(conn net.Conn, manifest *Manifest) { + msg := &ProtocolMessage{ + Type: "manifest", + Files: manifest.ToJSON(), + } + data, err := EncodeMessage(msg) + if err != nil { + return + } + data = append(data, '\n') + conn.Write(data) + + done := &ProtocolMessage{Type: "manifest_done"} + doneData, _ := EncodeMessage(done) + doneData = append(doneData, '\n') + conn.Write(doneData) +} + +func sendEvent(conn net.Conn, evt FileEvent) { + msg := &ProtocolMessage{ + Type: "event", + Path: evt.Path, + } + switch evt.Op { + case "create": + msg.Type = "event_create" + case "write": + msg.Type = "event_write" + case "remove": + msg.Type = "event_remove" + } + data, err := EncodeMessage(msg) + if err != nil { + return + } + data = append(data, '\n') + conn.Write(data) +} + func addWatchersRecursively(watcher *fsnotify.Watcher, root string, stack *GitignoreStack) error { return filepath.Walk(root, func(path string, info os.FileInfo, err error) error { if err != nil { @@ -1,6 +1,7 @@ package main import ( + "encoding/json" "os" "time" ) @@ -18,6 +19,59 @@ type FileMeta struct { Symlink string } +type FileMetaJSON struct { + Size int64 `json:"size"` + Mode uint32 `json:"mode"` + ModTime string `json:"modtime"` + Symlink string `json:"symlink,omitempty"` +} + type Manifest struct { Files map[string]FileMeta } + +type ProtocolMessage struct { + Type string `json:"type"` + Files map[string]FileMetaJSON `json:"files,omitempty"` + Path string `json:"path,omitempty"` + Content []byte `json:"content,omitempty"` +} + +func (m *Manifest) ToJSON() map[string]FileMetaJSON { + files := make(map[string]FileMetaJSON, len(m.Files)) + for k, v := range m.Files { + files[k] = FileMetaJSON{ + Size: v.Size, + Mode: uint32(v.Mode), + ModTime: v.ModTime.UTC().Format(time.RFC3339Nano), + Symlink: v.Symlink, + } + } + return files +} + +func ManifestFromJSON(files map[string]FileMetaJSON) *Manifest { + m := &Manifest{Files: make(map[string]FileMeta, len(files))} + for k, v := range files { + t, _ := time.Parse(time.RFC3339Nano, v.ModTime) + m.Files[k] = FileMeta{ + Size: v.Size, + Mode: os.FileMode(v.Mode), + ModTime: t, + Symlink: v.Symlink, + } + } + return m +} + +func DecodeMessage(data []byte) (*ProtocolMessage, error) { + var msg ProtocolMessage + if err := json.Unmarshal(data, &msg); err != nil { + return nil, err + } + return &msg, nil +} + +func EncodeMessage(msg *ProtocolMessage) ([]byte, error) { + return json.Marshal(msg) +} |