| 1 | //go:build !windows |
| 2 | |
| 3 | package skillwatch |
| 4 | |
| 5 | import ( |
| 6 | "errors" |
| 7 | "path/filepath" |
| 8 | "sync" |
| 9 | |
| 10 | "github.com/fsnotify/fsnotify" |
| 11 | ) |
| 12 | |
| 13 | var errBackendClosed = errors.New("watch backend closed") |
| 14 | |
| 15 | // nativeBackend watches directories in process. Add/Remove/Close run on one |
| 16 | // serialized registration goroutine while a pump drains Events and Errors, so |
| 17 | // a registration can never wait on fsnotify's error sender — the invariant the |
| 18 | // retired Store-owned watcher established. Physical watches are refcounted |
| 19 | // across registrations: the same directory is registered with the kernel once. |
| 20 | type nativeBackend struct { |
| 21 | svc *Service |
| 22 | |
| 23 | mu sync.Mutex |
| 24 | closed bool |
| 25 | refs map[string]int |
| 26 | watchers map[string]map[regKey]struct{} |
| 27 | byID map[uint64]regKey |
| 28 | |
| 29 | ops chan func() |
| 30 | pumpDone chan struct{} |
| 31 | watcher *fsnotify.Watcher |
| 32 | } |
| 33 | |
| 34 | type regKey struct { |
| 35 | id uint64 |
| 36 | gen uint64 |
| 37 | } |
| 38 | |
| 39 | var _ backend = (*nativeBackend)(nil) |
| 40 | |
| 41 | // newPlatformBackend selects the in-process native backend. ForceHelper exists |
| 42 | // so tests exercise the helper protocol on any OS. |
| 43 | func newPlatformBackend(svc *Service, opts Options) (backend, string, *helperClient) { |
| 44 | if opts.ForceHelper { |
| 45 | start := opts.HelperCommand |
| 46 | if start == nil { |
| 47 | start = defaultHelperCommand |
| 48 | } |
| 49 | h := newHelperClient(start, svc) |
| 50 | return h, "helper", h |
| 51 | } |
| 52 | return newNativeBackend(svc), "native", nil |
| 53 | } |
| 54 | |
| 55 | func newNativeBackend(svc *Service) *nativeBackend { |
| 56 | b := &nativeBackend{ |
| 57 | svc: svc, |
| 58 | refs: map[string]int{}, |
| 59 | watchers: map[string]map[regKey]struct{}{}, |
| 60 | byID: map[uint64]regKey{}, |
| 61 | ops: make(chan func(), 4096), |
| 62 | pumpDone: make(chan struct{}), |
| 63 | } |
| 64 | watcher, err := fsnotify.NewWatcher() |
| 65 | if err != nil { |
| 66 | // Without a backend every registration fails and degrades to the |
| 67 | // bounded scan fallback. |
| 68 | close(b.pumpDone) |
| 69 | return b |
| 70 | } |
| 71 | b.watcher = watcher |
| 72 | go b.registrationLoop() |
| 73 | go b.pump() |
| 74 | return b |
| 75 | } |
| 76 | |
| 77 | func (b *nativeBackend) registrationLoop() { |
| 78 | defer close(b.pumpDone) |
| 79 | for op := range b.ops { |
| 80 | op() |
| 81 | } |
| 82 | if b.watcher != nil { |
| 83 | _ = b.watcher.Close() |
| 84 | } |
| 85 | } |
| 86 | |
| 87 | func (b *nativeBackend) pump() { |
| 88 | events := b.watcher.Events |
| 89 | errors := b.watcher.Errors |
| 90 | for events != nil || errors != nil { |
| 91 | select { |
| 92 | case event, ok := <-events: |
| 93 | if !ok { |
| 94 | events = nil |
| 95 | continue |
| 96 | } |
| 97 | b.dispatch(event) |
| 98 | case _, ok := <-errors: |
| 99 | if !ok { |
| 100 | errors = nil |
| 101 | continue |
| 102 | } |
| 103 | // Backend-level failure: touch every covered registration so each |
| 104 | // root revalidates (and re-arms) through its subscribers. |
| 105 | b.mu.Lock() |
| 106 | keys := make([]regKey, 0, len(b.byID)) |
| 107 | for _, key := range b.byID { |
| 108 | keys = append(keys, key) |
| 109 | } |
| 110 | b.mu.Unlock() |
| 111 | for _, key := range keys { |
| 112 | b.svc.eventArrived(key.id, key.gen, OpWrite) |
| 113 | } |
| 114 | } |
| 115 | } |
| 116 | } |
| 117 | |
| 118 | func (b *nativeBackend) dispatch(event fsnotify.Event) { |
| 119 | var op Op |
| 120 | switch { |
| 121 | case event.Op&fsnotify.Create != 0: |
| 122 | op = OpCreate |
| 123 | case event.Op&fsnotify.Remove != 0: |
| 124 | op = OpRemove |
| 125 | case event.Op&fsnotify.Rename != 0: |
| 126 | op = OpRename |
| 127 | case event.Op&fsnotify.Write != 0: |
| 128 | op = OpWrite |
| 129 | case event.Op&fsnotify.Chmod != 0: |
| 130 | op = OpChmod |
| 131 | default: |
| 132 | return |
| 133 | } |
| 134 | b.mu.Lock() |
| 135 | // fsnotify reports the changed path (file or directory); registrations |
| 136 | // cover directories. Walk up to find the watched ancestor(s). |
| 137 | var keys []regKey |
| 138 | for dir := event.Name; ; { |
| 139 | if subs := b.watchers[dir]; subs != nil { |
| 140 | for key := range subs { |
| 141 | keys = append(keys, key) |
| 142 | } |
| 143 | } |
| 144 | parent := filepath.Dir(dir) |
| 145 | if parent == dir { |
| 146 | break |
| 147 | } |
| 148 | dir = parent |
| 149 | } |
| 150 | b.mu.Unlock() |
| 151 | for _, key := range keys { |
| 152 | b.svc.eventArrived(key.id, key.gen, op) |
| 153 | } |
| 154 | } |
| 155 | |
| 156 | func (b *nativeBackend) register(id, rootGen uint64, _ string, dirs []string) error { |
| 157 | b.mu.Lock() |
| 158 | watcher := b.watcher |
| 159 | b.mu.Unlock() |
| 160 | if watcher == nil { |
| 161 | return errBackendClosed |
| 162 | } |
| 163 | key := regKey{id: id, gen: rootGen} |
| 164 | result := make(chan error, 1) |
| 165 | ok := b.enqueue(func() { |
| 166 | // One critical section for Add and bookkeeping: physicalWatches>0 |
| 167 | // therefore implies the kernel watch is armed, and a dispatched event |
| 168 | // can never fall between "watching" and "mapped". |
| 169 | b.mu.Lock() |
| 170 | if b.closed { |
| 171 | b.mu.Unlock() |
| 172 | result <- errBackendClosed |
| 173 | return |
| 174 | } |
| 175 | var added []string |
| 176 | for _, dir := range dirs { |
| 177 | if b.refs[dir] > 0 { |
| 178 | // Already watched for another registration: just join it. |
| 179 | } else if err := watcher.Add(dir); err != nil { |
| 180 | for _, undo := range added { |
| 181 | b.removeRegistrationDirsLocked(key, undo) |
| 182 | } |
| 183 | b.mu.Unlock() |
| 184 | result <- err |
| 185 | return |
| 186 | } |
| 187 | b.refs[dir]++ |
| 188 | if b.watchers[dir] == nil { |
| 189 | b.watchers[dir] = map[regKey]struct{}{} |
| 190 | } |
| 191 | b.watchers[dir][key] = struct{}{} |
| 192 | added = append(added, dir) |
| 193 | } |
| 194 | b.byID[id] = key |
| 195 | b.mu.Unlock() |
| 196 | result <- nil |
| 197 | }) |
| 198 | if !ok { |
| 199 | return errBackendClosed |
| 200 | } |
| 201 | return <-result |
| 202 | } |
| 203 | |
| 204 | // removeRegistrationDirsLocked unrolls one registration's state for one dir |
| 205 | // and removes the kernel watch when its refcount reaches zero. |
| 206 | func (b *nativeBackend) removeRegistrationDirsLocked(key regKey, dir string) { |
| 207 | if subs := b.watchers[dir]; subs != nil { |
| 208 | if _, ok := subs[key]; ok { |
| 209 | delete(subs, key) |
| 210 | if len(subs) == 0 { |
| 211 | delete(b.watchers, dir) |
| 212 | } |
| 213 | b.refs[dir]-- |
| 214 | if b.refs[dir] <= 0 { |
| 215 | delete(b.refs, dir) |
| 216 | if b.watcher != nil { |
| 217 | _ = b.watcher.Remove(dir) |
| 218 | } |
| 219 | } |
| 220 | } |
| 221 | } |
| 222 | } |
| 223 | |
| 224 | // enqueue submits one serialized watcher operation. It reports false when the |
| 225 | // backend already closed or the queue is saturated; both surface as backend |
| 226 | // failure and degrade the root instead of blocking callers. Sends happen under |
| 227 | // mu, the same lock close() holds while closing the channel, so a late |
| 228 | // operation can never panic — and buffered sends never deadlock. |
| 229 | func (b *nativeBackend) enqueue(op func()) bool { |
| 230 | b.mu.Lock() |
| 231 | defer b.mu.Unlock() |
| 232 | if b.closed { |
| 233 | return false |
| 234 | } |
| 235 | select { |
| 236 | case b.ops <- op: |
| 237 | return true |
| 238 | default: |
| 239 | return false |
| 240 | } |
| 241 | } |
| 242 | |
| 243 | // removeRegistrationLocked unrolls one registration's map state and removes |
| 244 | // kernel watches whose refcount reached zero. Called only from the ops |
| 245 | // goroutine, which is what owns watcher mutation. |
| 246 | func (b *nativeBackend) removeRegistrationLocked(key regKey) { |
| 247 | if _, ok := b.byID[key.id]; !ok { |
| 248 | return |
| 249 | } |
| 250 | delete(b.byID, key.id) |
| 251 | for dir := range b.refs { |
| 252 | b.removeRegistrationDirsLocked(key, dir) |
| 253 | } |
| 254 | } |
| 255 | |
| 256 | func (b *nativeBackend) cancel(id uint64) { |
| 257 | b.mu.Lock() |
| 258 | if b.closed { |
| 259 | b.mu.Unlock() |
| 260 | return |
| 261 | } |
| 262 | key, ok := b.byID[id] |
| 263 | b.mu.Unlock() |
| 264 | if !ok { |
| 265 | return |
| 266 | } |
| 267 | done := make(chan struct{}) |
| 268 | b.enqueue(func() { |
| 269 | defer close(done) |
| 270 | b.mu.Lock() |
| 271 | b.removeRegistrationLocked(key) |
| 272 | b.mu.Unlock() |
| 273 | }) |
| 274 | <-done |
| 275 | } |
| 276 | |
| 277 | func (b *nativeBackend) close() error { |
| 278 | b.mu.Lock() |
| 279 | if b.closed { |
| 280 | b.mu.Unlock() |
| 281 | return nil |
| 282 | } |
| 283 | b.closed = true |
| 284 | close(b.ops) |
| 285 | b.mu.Unlock() |
| 286 | <-b.pumpDone |
| 287 | return nil |
| 288 | } |
| 289 | |
| 290 | func (b *nativeBackend) physicalWatches() uint64 { |
| 291 | b.mu.Lock() |
| 292 | defer b.mu.Unlock() |
| 293 | return uint64(len(b.refs)) |
| 294 | } |
| 295 |