| 1 | package hostfiles |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "path" |
| 9 | "strings" |
| 10 | "time" |
| 11 | ) |
| 12 | |
| 13 | type Options struct { |
| 14 | Deadline time.Duration // first unpublished modification to publish |
| 15 | Scan time.Duration // repair scan while every directory is watched |
| 16 | Poll time.Duration // full-scan period while watch coverage is missing |
| 17 | Skip func(rel string, isDir bool) bool |
| 18 | OnStats func(SyncStats) // called after each full scan; may be nil |
| 19 | } |
| 20 | |
| 21 | // DefaultOptions are region HA spec 1 §5.6's values. |
| 22 | func DefaultOptions() Options { |
| 23 | return Options{Deadline: time.Second, Scan: 10 * time.Minute, Poll: time.Second} |
| 24 | } |
| 25 | |
| 26 | type event struct { |
| 27 | rel string |
| 28 | rescan bool |
| 29 | } |
| 30 | |
| 31 | // watcher reports changes under one directory. Its event channel closes |
| 32 | // when it can report nothing more. |
| 33 | type watcher interface { |
| 34 | Events() <-chan event |
| 35 | Degraded() bool // some directory may have no working watch |
| 36 | Repair() error // restore full coverage, or say why not |
| 37 | Close() error |
| 38 | } |
| 39 | |
| 40 | // Run publishes dir's changes until ctx ends (it returns ctx.Err()), |
| 41 | // another writer supersedes the manifest (ErrSuperseded), or the manifest |
| 42 | // holds a path past the cap so no commit can succeed (ErrTooDeep). Where |
| 43 | // there is no watcher, or the watcher fails, it polls. |
| 44 | func Run(ctx context.Context, p *Publisher, dir string, o Options) error { |
| 45 | w, err := newWatcher(dir, o.Skip) |
| 46 | if err != nil { |
| 47 | log.Printf("hostfiles: no watcher for %s (%v): polling every %s", dir, err, o.Poll) |
| 48 | return run(ctx, p, dir, o, nil) |
| 49 | } |
| 50 | defer w.Close() |
| 51 | return run(ctx, p, dir, o, w) |
| 52 | } |
| 53 | |
| 54 | func run(ctx context.Context, p *Publisher, dir string, o Options, w watcher) error { |
| 55 | def := DefaultOptions() |
| 56 | if o.Deadline <= 0 { |
| 57 | o.Deadline = def.Deadline |
| 58 | } |
| 59 | if o.Scan <= 0 { |
| 60 | o.Scan = def.Scan |
| 61 | } |
| 62 | if o.Poll <= 0 { |
| 63 | o.Poll = def.Poll |
| 64 | } |
| 65 | root, err := openTree(dir) |
| 66 | if err != nil { |
| 67 | return fmt.Errorf("hostfiles: %s: %w", dir, err) |
| 68 | } |
| 69 | defer root.Close() |
| 70 | |
| 71 | var events <-chan event |
| 72 | if w != nil { |
| 73 | events = w.Events() |
| 74 | } |
| 75 | q := newQueue(o.Deadline) |
| 76 | tick := time.NewTicker(o.Deadline / 4) |
| 77 | defer tick.Stop() |
| 78 | |
| 79 | needScan := true // the first pass is a full scan: changes made while nothing watched |
| 80 | var lastScan, lastScanTry, lastRepair time.Time |
| 81 | coverageLost, scanFailing := false, false // for logging on transitions only |
| 82 | loggedTooDeep := false // publishOne's over-cap skip, logged once |
| 83 | for { |
| 84 | select { |
| 85 | case <-ctx.Done(): |
| 86 | return ctx.Err() |
| 87 | case ev, open := <-events: |
| 88 | if !open { |
| 89 | // The watcher has stopped for good. A nil channel never |
| 90 | // fires, and with no watcher every pass below polls. |
| 91 | log.Printf("hostfiles: watcher for %s stopped: polling every %s", dir, o.Poll) |
| 92 | events, w = nil, nil |
| 93 | continue |
| 94 | } |
| 95 | if ev.rescan { |
| 96 | needScan = true |
| 97 | } else { |
| 98 | q.Touch(ev.rel, time.Now()) |
| 99 | } |
| 100 | continue |
| 101 | case <-tick.C: |
| 102 | } |
| 103 | now := time.Now() |
| 104 | if w != nil && w.Degraded() { |
| 105 | if !coverageLost { |
| 106 | coverageLost = true |
| 107 | log.Printf("hostfiles: %s: watch coverage is missing: polling every %s until it is restored", dir, o.Poll) |
| 108 | } |
| 109 | // A repair is a full tree walk: try it once per polling period. |
| 110 | if now.Sub(lastRepair) >= o.Poll { |
| 111 | lastRepair = now |
| 112 | if err := w.Repair(); err == nil { |
| 113 | needScan = true // whatever changed while coverage was missing |
| 114 | } |
| 115 | } |
| 116 | } |
| 117 | polling := w == nil || w.Degraded() |
| 118 | if coverageLost && !polling { |
| 119 | coverageLost = false |
| 120 | log.Printf("hostfiles: %s: watch coverage restored", dir) |
| 121 | } |
| 122 | switch { |
| 123 | case polling && now.Sub(lastScan) >= o.Poll: |
| 124 | needScan = true |
| 125 | case !polling && now.Sub(lastScan) >= o.Scan: |
| 126 | needScan = true |
| 127 | } |
| 128 | if needScan && (!scanFailing || now.Sub(lastScanTry) >= o.Poll) { |
| 129 | lastScanTry = now |
| 130 | st, err := syncRoot(ctx, p, root, o.Skip) |
| 131 | if err != nil { |
| 132 | if ctx.Err() != nil { |
| 133 | return ctx.Err() |
| 134 | } |
| 135 | if !scanFailing { |
| 136 | scanFailing = true |
| 137 | log.Printf("hostfiles: %s: scan failing, retrying every %s: %v", dir, o.Poll, err) |
| 138 | } |
| 139 | } else { |
| 140 | if scanFailing { |
| 141 | scanFailing = false |
| 142 | log.Printf("hostfiles: %s: scan recovered", dir) |
| 143 | } |
| 144 | needScan, lastScan = false, now |
| 145 | if o.OnStats != nil { |
| 146 | o.OnStats(st) |
| 147 | } |
| 148 | } |
| 149 | } |
| 150 | for _, rel := range q.Due(now) { |
| 151 | again, err := publishOne(ctx, p, root, rel, o.Skip) |
| 152 | if errors.Is(err, ErrTooDeep) { |
| 153 | // Past the path cap: never staged, so never in a manifest a |
| 154 | // reader would refuse, and not retried (it cannot shrink). |
| 155 | // Logged once per run, so neither the log nor a set of |
| 156 | // paths grows with a tenant's supply of them. |
| 157 | if !loggedTooDeep { |
| 158 | loggedTooDeep = true |
| 159 | log.Printf("hostfiles: %s: a path past the %d-byte cap is not published (further ones are skipped without a line): %v", dir, MaxPathBytes, err) |
| 160 | } |
| 161 | continue |
| 162 | } |
| 163 | if err != nil { |
| 164 | log.Printf("hostfiles: %s: %v", rel, err) |
| 165 | q.Touch(rel, now) // try again at the next deadline |
| 166 | continue |
| 167 | } |
| 168 | if again { |
| 169 | q.Touch(rel, now) |
| 170 | } |
| 171 | } |
| 172 | if p.Dirty() { |
| 173 | if _, err := p.Commit(ctx); errors.Is(err, ErrSuperseded) { |
| 174 | return err |
| 175 | } else if errors.Is(err, ErrTooDeep) { |
| 176 | // The manifest holds a path past the cap: no commit can |
| 177 | // ever succeed, so retrying would only spin. One line, and |
| 178 | // stop. |
| 179 | log.Printf("hostfiles: %s: cannot commit, stopping: %v", dir, err) |
| 180 | return err |
| 181 | } else if err != nil { |
| 182 | log.Printf("hostfiles: commit: %v", err) // staged changes stay; the next tick retries |
| 183 | } |
| 184 | } |
| 185 | } |
| 186 | } |
| 187 | |
| 188 | // publishOne stages one path's current state. again is true when the |
| 189 | // caller should read the path again at the next deadline: the file changed |
| 190 | // while it was read (its image is published, flagged), or it was replaced |
| 191 | // mid-look and nothing was read. |
| 192 | // |
| 193 | // A tombstone is staged only on proof: the name is gone, or something |
| 194 | // that is not a regular file now stands there or above it. A file that |
| 195 | // exists but could not be read is an error, and its manifest entry stays. |
| 196 | func publishOne(ctx context.Context, p *Publisher, root *dirHandle, rel string, skip func(string, bool) bool) (again bool, err error) { |
| 197 | clean, err := CleanPath(rel) |
| 198 | if err != nil { |
| 199 | return false, nil // a path the watcher should never have produced: skip it |
| 200 | } |
| 201 | if err := withinTreeCap(clean); err != nil { |
| 202 | return false, err // the caller skips it for good: no reader would accept it |
| 203 | } |
| 204 | if skipped(skip, clean, false) { |
| 205 | return false, nil |
| 206 | } |
| 207 | parent, name := path.Split(clean) |
| 208 | dir, gone, err := root.descend(strings.TrimSuffix(parent, "/")) |
| 209 | if err != nil { |
| 210 | return false, err |
| 211 | } |
| 212 | if gone { |
| 213 | return false, p.StageDelete(clean) |
| 214 | } |
| 215 | if dir != root { |
| 216 | defer dir.Close() |
| 217 | } |
| 218 | body, mode, state, err := readStable(dir, name) |
| 219 | if err != nil { |
| 220 | return false, err |
| 221 | } |
| 222 | switch state { |
| 223 | case fileGone, fileNotRegular: |
| 224 | // StageDelete is a no-op for a path the manifest never had, which |
| 225 | // is what a directory's own event is. |
| 226 | return false, p.StageDelete(clean) |
| 227 | case fileUnreadable: |
| 228 | return false, fmt.Errorf("%s exists but could not be read; keeping what is published", clean) |
| 229 | case fileRetry: |
| 230 | return true, nil |
| 231 | case fileTooLarge: |
| 232 | return false, p.StageSkipped(clean, "too-large") |
| 233 | } |
| 234 | return state == fileTorn, p.Stage(ctx, clean, body, mode, state == fileTorn) |
| 235 | } |
| 236 | |
amadan