| 1 | package skillwatch |
| 2 | |
| 3 | import ( |
| 4 | "io" |
| 5 | "os" |
| 6 | "path/filepath" |
| 7 | "slices" |
| 8 | "sync" |
| 9 | |
| 10 | "github.com/fsnotify/fsnotify" |
| 11 | ) |
| 12 | |
| 13 | // MaybeRunHelper is the internal entry hosts call before any other work. When |
| 14 | // the process was started with the watcher helper environment flag it serves |
| 15 | // the pipe protocol on stdin/stdout and reports true; the caller must exit |
| 16 | // without initializing the application. |
| 17 | func MaybeRunHelper() bool { |
| 18 | if os.Getenv(watchHelperEnv) != "1" { |
| 19 | return false |
| 20 | } |
| 21 | _ = RunHelper(os.Stdin, os.Stdout) |
| 22 | return true |
| 23 | } |
| 24 | |
| 25 | // RunHelper serves one host connection: control frames are processed strictly |
| 26 | // serially on this goroutine (all watcher Add/Remove/Close calls happen here, |
| 27 | // which is the whole reason Windows watching moved into a helper), while a |
| 28 | // pump forwards filesystem events and a writer goroutine serializes outbound |
| 29 | // frames. Any write failure terminates the helper; the host detects the closed |
| 30 | // pipe and applies its restart budget. |
| 31 | func RunHelper(r io.Reader, w io.Writer) error { |
| 32 | watcher, err := fsnotify.NewWatcher() |
| 33 | if err != nil { |
| 34 | _ = writeFrame(w, frame{Kind: wireError, Msg: "backend unavailable: " + err.Error()}) |
| 35 | return err |
| 36 | } |
| 37 | |
| 38 | outbound := make(chan frame, 256) |
| 39 | writerDone := make(chan struct{}) |
| 40 | go func() { |
| 41 | defer close(writerDone) |
| 42 | for f := range outbound { |
| 43 | if err := writeFrame(w, f); err != nil { |
| 44 | return |
| 45 | } |
| 46 | } |
| 47 | }() |
| 48 | |
| 49 | var mu sync.Mutex |
| 50 | // Physical directories are watched once across logical registrations. |
| 51 | dirRefs := map[string]int{} |
| 52 | // These indexes fence generations and map paths back to registrations. |
| 53 | regDirs := map[uint64]map[string]struct{}{} |
| 54 | regGen := map[uint64]uint64{} |
| 55 | pathRegs := map[string]map[uint64]struct{}{} |
| 56 | |
| 57 | pumpDone := make(chan struct{}) |
| 58 | go func() { |
| 59 | defer close(pumpDone) |
| 60 | for { |
| 61 | select { |
| 62 | case event, ok := <-watcher.Events: |
| 63 | if !ok { |
| 64 | return |
| 65 | } |
| 66 | var op Op |
| 67 | switch { |
| 68 | case event.Op&fsnotify.Create != 0: |
| 69 | op = OpCreate |
| 70 | case event.Op&fsnotify.Remove != 0: |
| 71 | op = OpRemove |
| 72 | case event.Op&fsnotify.Rename != 0: |
| 73 | op = OpRename |
| 74 | case event.Op&fsnotify.Write != 0: |
| 75 | op = OpWrite |
| 76 | case event.Op&fsnotify.Chmod != 0: |
| 77 | op = OpChmod |
| 78 | default: |
| 79 | continue |
| 80 | } |
| 81 | mu.Lock() |
| 82 | // fsnotify reports the changed path; registrations cover |
| 83 | // directories. Walk up to the watched ancestor(s). |
| 84 | var ids []uint64 |
| 85 | for dir := event.Name; ; { |
| 86 | for id := range pathRegs[dir] { |
| 87 | ids = append(ids, id) |
| 88 | } |
| 89 | parent := filepath.Dir(dir) |
| 90 | if parent == dir { |
| 91 | break |
| 92 | } |
| 93 | dir = parent |
| 94 | } |
| 95 | mu.Unlock() |
| 96 | slices.Sort(ids) |
| 97 | for _, id := range ids { |
| 98 | mu.Lock() |
| 99 | gen := regGen[id] |
| 100 | mu.Unlock() |
| 101 | select { |
| 102 | case outbound <- frame{Kind: wireEvent, ID: id, RootGen: gen, Op: op}: |
| 103 | default: |
| 104 | // The host timeout detects a stalled drain. |
| 105 | } |
| 106 | } |
| 107 | case _, ok := <-watcher.Errors: |
| 108 | if !ok { |
| 109 | return |
| 110 | } |
| 111 | select { |
| 112 | case outbound <- frame{Kind: wireError, Msg: "watcher backend error"}: |
| 113 | default: |
| 114 | } |
| 115 | } |
| 116 | } |
| 117 | }() |
| 118 | |
| 119 | defer func() { |
| 120 | // Stop pump sources before draining the writer. |
| 121 | _ = watcher.Close() |
| 122 | <-pumpDone |
| 123 | close(outbound) |
| 124 | <-writerDone |
| 125 | }() |
| 126 | |
| 127 | _ = writeFrame(w, frame{Kind: wireReady}) |
| 128 | for { |
| 129 | f, err := readFrame(r) |
| 130 | if err != nil { |
| 131 | return err |
| 132 | } |
| 133 | switch f.Kind { |
| 134 | case wireRegister: |
| 135 | if err := helperRegister(watcher, &mu, dirRefs, regDirs, regGen, pathRegs, f); err != nil { |
| 136 | _ = writeFrame(w, frame{Kind: wireError, ID: f.ID, Msg: err.Error()}) |
| 137 | continue |
| 138 | } |
| 139 | _ = writeFrame(w, frame{Kind: wireRegistered, ID: f.ID}) |
| 140 | case wireCancel: |
| 141 | helperCancel(&mu, dirRefs, regDirs, regGen, pathRegs, watcher, f.ID) |
| 142 | case wirePing: |
| 143 | _ = writeFrame(w, frame{Kind: wirePong}) |
| 144 | case wireShutdown: |
| 145 | return nil |
| 146 | default: |
| 147 | _ = writeFrame(w, frame{Kind: wireError, ID: f.ID, Msg: "unexpected frame"}) |
| 148 | } |
| 149 | } |
| 150 | } |
| 151 | |
| 152 | func helperRegister(watcher *fsnotify.Watcher, mu *sync.Mutex, dirRefs map[string]int, regDirs map[uint64]map[string]struct{}, regGen map[uint64]uint64, pathRegs map[string]map[uint64]struct{}, f frame) error { |
| 153 | mu.Lock() |
| 154 | defer mu.Unlock() |
| 155 | added := make([]string, 0, len(f.Dirs)) |
| 156 | dirs := make(map[string]struct{}, len(f.Dirs)) |
| 157 | for _, dir := range f.Dirs { |
| 158 | clean := filepath.Clean(dir) |
| 159 | dirs[clean] = struct{}{} |
| 160 | if dirRefs[clean] == 0 { |
| 161 | // The only goroutine that touches the watcher: the serialized |
| 162 | // control loop, so Add cannot block a Close or another Add. |
| 163 | if err := watcher.Add(clean); err != nil { |
| 164 | for _, undo := range added { |
| 165 | helperDropDir(dirRefs, pathRegs, watcher, undo, f.ID) |
| 166 | } |
| 167 | return err |
| 168 | } |
| 169 | } |
| 170 | dirRefs[clean]++ |
| 171 | if pathRegs[clean] == nil { |
| 172 | pathRegs[clean] = map[uint64]struct{}{} |
| 173 | } |
| 174 | pathRegs[clean][f.ID] = struct{}{} |
| 175 | added = append(added, clean) |
| 176 | } |
| 177 | regDirs[f.ID] = dirs |
| 178 | regGen[f.ID] = f.RootGen |
| 179 | return nil |
| 180 | } |
| 181 | |
| 182 | func helperCancel(mu *sync.Mutex, dirRefs map[string]int, regDirs map[uint64]map[string]struct{}, regGen map[uint64]uint64, pathRegs map[string]map[uint64]struct{}, watcher *fsnotify.Watcher, id uint64) { |
| 183 | mu.Lock() |
| 184 | defer mu.Unlock() |
| 185 | for dir := range regDirs[id] { |
| 186 | helperDropDir(dirRefs, pathRegs, watcher, dir, id) |
| 187 | } |
| 188 | delete(regDirs, id) |
| 189 | delete(regGen, id) |
| 190 | } |
| 191 | |
| 192 | func helperDropDir(dirRefs map[string]int, pathRegs map[string]map[uint64]struct{}, watcher *fsnotify.Watcher, dir string, id uint64) { |
| 193 | // Caller holds mu. |
| 194 | if regs := pathRegs[dir]; regs != nil { |
| 195 | delete(regs, id) |
| 196 | if len(regs) == 0 { |
| 197 | delete(pathRegs, dir) |
| 198 | } |
| 199 | } |
| 200 | dirRefs[dir]-- |
| 201 | if dirRefs[dir] <= 0 { |
| 202 | delete(dirRefs, dir) |
| 203 | _ = watcher.Remove(dir) |
| 204 | } |
| 205 | } |
| 206 |