For reading and verification only. You may not copy, use, modify or redistribute this code. Full notice on the main page.
tunnel/feed.go
| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "sort" |
| 7 | "sync" |
| 8 | "time" |
| 9 | ) |
| 10 | |
| 11 | // The feed consumer (the design plan, sections 2, 3 and 5; the contract, section 4): what the node must stop serving, learned from |
| 12 | // GET /v1/tunnel/feed within policy timing.feed_apply_deadline_s. A node never decides to cut a device by itself; it cuts what the control plane says |
| 13 | // to cut, and it keeps serving what is already authenticated until valid_until when the control plane cannot be asked. |
| 14 | // |
| 15 | // revoke, release, license_end the device's control connection is closed and the streams of its host are killed |
| 16 | // pause no new stream; the streams in flight are killed; the connection stays (a resume arrives as a statement) and the |
| 17 | // node asks the control plane for the device's signed statement, which carries the pause, and forwards it |
| 18 | // resume the node asks again: the control plane's answer decides the status, and its statement (resumed) is forwarded |
| 19 | // kill, kill_clear the global switch: every connection is closed (CLOSE killed) and every stream killed, and nobody is served |
| 20 | // (control_unavailable) until it is cleared |
| 21 | // reset the cursor is not one the control plane can serve from: every tunnel is closed, the devices reconnect and are |
| 22 | // authorized again |
| 23 | // |
| 24 | // Events are applied in epoch order, each once: an event at or below the cursor is dropped, so a duplicate or a replay changes nothing, and a batch |
| 25 | // that arrives out of order is sorted first. An event older than a connection's last answer from the control plane (by the control plane's own clock) |
| 26 | // is already part of that answer and is skipped: after a restart the feed is replayed from the start, and a device that came back since its revoke |
| 27 | // must not be cut again by the old event. |
| 28 | |
| 29 | // feedWaitSlack is how much longer than its hold the node lets one poll take. |
| 30 | const feedWaitSlack = 10 * time.Second |
| 31 | |
| 32 | // startBackground starts the feed loop and the node's background tasks. The feed's first answer is awaited (bounded) so that the node does not accept |
| 33 | // a device before it knows what it must not serve. |
| 34 | func (s *Server) startBackground() { |
| 35 | for _, task := range append([]func(context.Context){s.openai.Run, s.sweepLoop}, s.cfg.Background...) { |
| 36 | s.bg.Add(1) |
| 37 | go func(task func(context.Context)) { |
| 38 | defer s.bg.Done() |
| 39 | task(s.ctx) |
| 40 | }(task) |
| 41 | } |
| 42 | if s.cfg.Feed == nil { |
| 43 | return |
| 44 | } |
| 45 | caughtUp := make(chan struct{}) |
| 46 | s.bg.Add(1) |
| 47 | go func() { |
| 48 | defer s.bg.Done() |
| 49 | s.runFeed(s.ctx, caughtUp) |
| 50 | }() |
| 51 | select { |
| 52 | case <-caughtUp: |
| 53 | case <-time.After(s.cfg.FeedStartWait): |
| 54 | s.log.Warn("feed_late", "waited_ms", s.cfg.FeedStartWait.Milliseconds()) |
| 55 | } |
| 56 | } |
| 57 | |
| 58 | // runFeed polls until ctx ends. close(caughtUp) runs once, after the first answer applied or the first failure. |
| 59 | func (s *Server) runFeed(ctx context.Context, caughtUp chan struct{}) { |
| 60 | var cursor int64 |
| 61 | failures := 0 |
| 62 | first := true |
| 63 | signal := func() { |
| 64 | if first { |
| 65 | first = false |
| 66 | close(caughtUp) |
| 67 | } |
| 68 | } |
| 69 | defer signal() |
| 70 | for ctx.Err() == nil { |
| 71 | started := s.now() |
| 72 | wait := s.cfg.FeedPoll |
| 73 | if first { |
| 74 | wait = 0 // the first look answers at once: it is the one the listeners wait for |
| 75 | } |
| 76 | pctx, cancel := context.WithTimeout(ctx, s.cfg.FeedPoll+feedWaitSlack) |
| 77 | answer, err := s.cfg.Feed.Feed(pctx, cursor, wait) |
| 78 | cancel() |
| 79 | if err != nil { |
| 80 | if ctx.Err() != nil { |
| 81 | return |
| 82 | } |
| 83 | s.health.record(s.now(), err) |
| 84 | failures++ |
| 85 | s.log.Warn("feed_failed", "reason", errorWord(err), "failures", failures) |
| 86 | signal() |
| 87 | // Retry quickly (never beyond the poll period: the moment the control plane is back, a revoke is heard within the apply deadline). |
| 88 | if !s.clk.sleep(ctx, jitter(min(backoff(failures), s.cfg.FeedPoll), 20)) { |
| 89 | return |
| 90 | } |
| 91 | continue |
| 92 | } |
| 93 | failures = 0 |
| 94 | s.health.good(s.now()) |
| 95 | next, applied := s.applyFeed(answer, cursor) |
| 96 | cursor = next |
| 97 | signal() |
| 98 | if !applied && wait > 0 { |
| 99 | // Nothing new in a poll that asked to be held. A control plane that holds it has already spent the period; one that answers at once is |
| 100 | // polled once per period. (The first look asks for no hold: the long polls start right after it.) |
| 101 | if rest := wait - s.now().Sub(started); rest > 0 && !s.clk.sleep(ctx, rest) { |
| 102 | return |
| 103 | } |
| 104 | } |
| 105 | } |
| 106 | } |
| 107 | |
| 108 | // applyFeed applies one answer and returns the cursor for the next poll, and whether the answer carried anything the node acted on. |
| 109 | func (s *Server) applyFeed(answer FeedAnswer, cursor int64) (int64, bool) { |
| 110 | if answer.Skipped > 0 { |
| 111 | s.log.Warn("feed_events_skipped", "count", answer.Skipped) |
| 112 | } |
| 113 | if answer.Reset { |
| 114 | s.log.Warn("feed_reset", "cursor", cursor, "epoch", answer.Epoch) |
| 115 | s.closeAll(closeExpired) |
| 116 | return answer.Epoch, true |
| 117 | } |
| 118 | if answer.Epoch < cursor { |
| 119 | // The control plane's feed never goes back without saying reset; an answer that does is not one this node can act on. |
| 120 | s.log.Warn("feed_behind_cursor", "cursor", cursor, "epoch", answer.Epoch) |
| 121 | return cursor, false |
| 122 | } |
| 123 | events := append([]FeedEvent(nil), answer.Events...) |
| 124 | sort.SliceStable(events, func(i, j int) bool { return events[i].Epoch < events[j].Epoch }) |
| 125 | next := cursor |
| 126 | applied := false |
| 127 | for _, event := range events { |
| 128 | if event.Epoch <= next { |
| 129 | continue // a duplicate, or an event this node has already applied |
| 130 | } |
| 131 | s.applyEvent(event) |
| 132 | next, applied = event.Epoch, true |
| 133 | } |
| 134 | return max(next, answer.Epoch), applied |
| 135 | } |
| 136 | |
| 137 | func (s *Server) applyEvent(event FeedEvent) { |
| 138 | s.log.Info("feed_event", "type", event.Type, "epoch", event.Epoch) |
| 139 | s.met.feedEvents.inc(event.Type) |
| 140 | // Noted before the table is looked at: a HELLO that joins the table after that look has the note to read, and one that joined before it is found. |
| 141 | s.noteEvent(event) |
| 142 | switch event.Type { |
| 143 | case feedKill: |
| 144 | s.setKilled(true) |
| 145 | return |
| 146 | case feedKillClear: |
| 147 | if s.killed.Swap(false) { |
| 148 | s.log.Warn("kill_switch_cleared") |
| 149 | } |
| 150 | return |
| 151 | } |
| 152 | reason := map[string]string{feedRevoke: closeRevoked, feedRelease: closeReleased, feedLicenseEnd: closeRevoked}[event.Type] |
| 153 | hosts := s.routes.hostsOfLabel(event.Label) |
| 154 | for _, cc := range s.routes.controls(event.Label) { |
| 155 | if cc.answeredAfter(event.At) { |
| 156 | continue // the answer the control plane gave this connection already knew |
| 157 | } |
| 158 | switch event.Type { |
| 159 | case feedPause: |
| 160 | cc.pause() |
| 161 | s.routes.killStreams(cc.hostName(), closePaused, true) |
| 162 | s.refreshAsync(cc) |
| 163 | case feedResume: |
| 164 | s.refreshAsync(cc) |
| 165 | default: |
| 166 | cc.close(reason) |
| 167 | s.routes.killStreams(cc.hostName(), reason, true) |
| 168 | } |
| 169 | } |
| 170 | if reason != "" { |
| 171 | // Streams of a host whose control connection is already gone (the device dropped, the streams did not yet). |
| 172 | for _, host := range hosts { |
| 173 | s.routes.killStreams(host, reason, false) |
| 174 | } |
| 175 | } |
| 176 | } |
| 177 | |
| 178 | // ---------------------------------------------------------------------------------------------------------------------------- events during authorize |
| 179 | |
| 180 | // feedWatch collects the feed events that are applied while one exchange waits for the control plane. The answer it gets was decided before them, so |
| 181 | // it cannot be served as it is: a revoke applied meanwhile found no connection to close (a HELLO is not in the routing table until its OK), and a |
| 182 | // pause applied meanwhile to a connection that is renewing is written over by the answer's status. The notes live exactly as long as the call: nothing |
| 183 | // is kept for an exchange that is not waiting. |
| 184 | type feedWatch struct { |
| 185 | s *Server |
| 186 | mu sync.Mutex |
| 187 | notes []FeedEvent |
| 188 | } |
| 189 | |
| 190 | func (s *Server) watchFeed() *feedWatch { |
| 191 | w := &feedWatch{s: s} |
| 192 | s.watchMu.Lock() |
| 193 | s.watches[w] = struct{}{} |
| 194 | s.watchMu.Unlock() |
| 195 | return w |
| 196 | } |
| 197 | |
| 198 | func (w *feedWatch) stop() { |
| 199 | w.s.watchMu.Lock() |
| 200 | delete(w.s.watches, w) |
| 201 | w.s.watchMu.Unlock() |
| 202 | } |
| 203 | |
| 204 | // noteEvent hands an applied event to every exchange that is waiting for authorize. The kill switch and a reset are not among them: they close every |
| 205 | // connection, the exchanges in flight included. |
| 206 | func (s *Server) noteEvent(event FeedEvent) { |
| 207 | switch event.Type { |
| 208 | case feedRevoke, feedRelease, feedLicenseEnd, feedPause, feedResume: |
| 209 | default: |
| 210 | return |
| 211 | } |
| 212 | s.watchMu.Lock() |
| 213 | defer s.watchMu.Unlock() |
| 214 | for w := range s.watches { |
| 215 | w.mu.Lock() |
| 216 | w.notes = append(w.notes, event) |
| 217 | w.mu.Unlock() |
| 218 | } |
| 219 | } |
| 220 | |
| 221 | // feedVerdict is what the events applied during a call mean for its answer. |
| 222 | type feedVerdict struct { |
| 223 | cut string // the answer must not be served: the CLOSE reason of the event that overtook it |
| 224 | err string // what a device that has not been told OK yet is answered |
| 225 | paused bool // the last pause or resume event of the host was a pause |
| 226 | refresh bool // a pause or a resume came: the control plane is asked again, its answer decides |
| 227 | } |
| 228 | |
| 229 | // verdict reads the notes that concern host, leaving out those older than the answer (by the control plane's own clock, as answeredAfter does: the |
| 230 | // answer already knew them). |
| 231 | func (w *feedWatch) verdict(host string, serverTime int64) feedVerdict { |
| 232 | w.mu.Lock() |
| 233 | defer w.mu.Unlock() |
| 234 | var v feedVerdict |
| 235 | for _, n := range w.notes { |
| 236 | if !hostHasLabel(host, n.Label) || serverTime > n.At { |
| 237 | continue |
| 238 | } |
| 239 | switch n.Type { |
| 240 | case feedRevoke, feedRelease: |
| 241 | if v.cut == "" { |
| 242 | v.cut, v.err = map[string]string{feedRevoke: closeRevoked, feedRelease: closeReleased}[n.Type], codeDeviceRevoked |
| 243 | } |
| 244 | case feedLicenseEnd: |
| 245 | if v.cut == "" { |
| 246 | v.cut, v.err = closeRevoked, codeLicenseInactive |
| 247 | } |
| 248 | case feedPause: |
| 249 | v.paused, v.refresh = true, true |
| 250 | case feedResume: |
| 251 | v.paused, v.refresh = false, true |
| 252 | } |
| 253 | } |
| 254 | return v |
| 255 | } |
| 256 | |
| 257 | // answeredAfter reports whether the control plane's last answer for this connection is newer than an event stamped at (epoch seconds). |
| 258 | func (c *ctrl) answeredAfter(at int64) bool { |
| 259 | c.mu.Lock() |
| 260 | defer c.mu.Unlock() |
| 261 | return c.authAt > at |
| 262 | } |
| 263 | |
| 264 | // setKilled turns the kill switch on: every control connection is closed (CLOSE killed) and every stream killed. While it is on no device is served. |
| 265 | func (s *Server) setKilled(on bool) { |
| 266 | if !on { |
| 267 | s.killed.Store(false) |
| 268 | return |
| 269 | } |
| 270 | if s.killed.Swap(true) { |
| 271 | return |
| 272 | } |
| 273 | s.log.Warn("kill_switch_on") |
| 274 | s.closeAll(closeKilled) |
| 275 | } |
| 276 | |
| 277 | // closeAll closes every control connection with reason and kills every client stream: the kill switch and a feed reset. |
| 278 | func (s *Server) closeAll(reason string) { |
| 279 | s.mu.Lock() |
| 280 | live := make([]*ctrl, 0, len(s.conns)) |
| 281 | for cc := range s.conns { |
| 282 | live = append(live, cc) |
| 283 | } |
| 284 | s.mu.Unlock() |
| 285 | for _, cc := range live { |
| 286 | cc.close(reason) |
| 287 | } |
| 288 | s.routes.killAllStreams(reason) |
| 289 | } |
| 290 | |
| 291 | // ---------------------------------------------------------------------------------------------------------------------------- refresh of a connection |
| 292 | |
| 293 | // refreshAsync asks the control plane again about one connection (after a pause or a resume event), in the background: the answer decides its status, |
| 294 | // and its signed statement (which carries the pause or the resume) is forwarded to the device. Events that arrive while a refresh runs are folded into |
| 295 | // one more. Until the control plane answers, a paused connection stays paused: the node never lifts a pause by itself. |
| 296 | func (s *Server) refreshAsync(cc *ctrl) { |
| 297 | cc.mu.Lock() |
| 298 | if cc.refreshing { |
| 299 | cc.refreshAgain = true |
| 300 | cc.mu.Unlock() |
| 301 | return |
| 302 | } |
| 303 | cc.refreshing = true |
| 304 | cc.mu.Unlock() |
| 305 | if !s.goBackground(func() { s.refreshLoop(cc) }) { |
| 306 | cc.mu.Lock() |
| 307 | cc.refreshing = false |
| 308 | cc.mu.Unlock() |
| 309 | } |
| 310 | } |
| 311 | |
| 312 | // goBackground runs task as one of the node's background tasks, unless the node has ended: the wait for them (shutdown) must not start while one is |
| 313 | // being added. |
| 314 | func (s *Server) goBackground(task func()) bool { |
| 315 | s.mu.Lock() |
| 316 | defer s.mu.Unlock() |
| 317 | if s.ctx.Err() != nil { |
| 318 | return false |
| 319 | } |
| 320 | s.bg.Add(1) |
| 321 | go func() { |
| 322 | defer s.bg.Done() |
| 323 | task() |
| 324 | }() |
| 325 | return true |
| 326 | } |
| 327 | |
| 328 | func (s *Server) refreshLoop(cc *ctrl) { |
| 329 | ctx, cancel := context.WithCancel(s.ctx) |
| 330 | defer cancel() |
| 331 | go func() { |
| 332 | select { |
| 333 | case <-cc.done: |
| 334 | cancel() |
| 335 | case <-ctx.Done(): |
| 336 | } |
| 337 | }() |
| 338 | failures := 0 |
| 339 | for ctx.Err() == nil { |
| 340 | cc.mu.Lock() |
| 341 | cc.refreshAgain = false |
| 342 | cc.mu.Unlock() |
| 343 | if err := s.refreshOnce(cc); err != nil { |
| 344 | failures++ |
| 345 | if !s.clk.sleep(ctx, jitter(backoff(failures), 20)) { |
| 346 | break |
| 347 | } |
| 348 | continue |
| 349 | } |
| 350 | failures = 0 |
| 351 | cc.mu.Lock() |
| 352 | again := cc.refreshAgain |
| 353 | if !again { |
| 354 | cc.refreshing = false |
| 355 | } |
| 356 | cc.mu.Unlock() |
| 357 | if !again { |
| 358 | return |
| 359 | } |
| 360 | } |
| 361 | cc.mu.Lock() |
| 362 | cc.refreshing = false |
| 363 | cc.mu.Unlock() |
| 364 | } |
| 365 | |
| 366 | var errRefreshLater = errors.New("the control plane did not answer") |
| 367 | |
| 368 | // refreshOnce is one authorize call for an established connection. It returns errRefreshLater when the control plane could not be asked (try again), |
| 369 | // and nil when the connection's state is settled (updated, or closed). |
| 370 | func (s *Server) refreshOnce(cc *ctrl) error { |
| 371 | cc.mu.Lock() |
| 372 | req := AuthorizeRequest{DeviceID: cc.sub, SessionJTI: cc.jti, Reason: "renew", Node: s.cfg.NodeID} |
| 373 | host := cc.host |
| 374 | cc.mu.Unlock() |
| 375 | if s.killed.Load() { |
| 376 | return nil |
| 377 | } |
| 378 | resp, err := s.authorize(req) |
| 379 | if err == nil { |
| 380 | if cerr := checkAuthorize(resp, req); cerr != nil { |
| 381 | s.health.bad(s.now()) |
| 382 | err = &ControlError{Status: 502, Code: "bad_response"} |
| 383 | } |
| 384 | } |
| 385 | if err != nil { |
| 386 | if isRefusal(err) { |
| 387 | // The control plane says this device is no longer served (revoked, license over, not enrolled): it is cut. |
| 388 | s.log.Info("refresh_refused", "conn", cc.id, "reason", errForControl(err)) |
| 389 | cc.close(closeRevoked) |
| 390 | s.routes.killStreams(host, closeRevoked, true) |
| 391 | return nil |
| 392 | } |
| 393 | return errRefreshLater |
| 394 | } |
| 395 | if resp.E2EKill { |
| 396 | s.setKilled(true) |
| 397 | return nil |
| 398 | } |
| 399 | if resp.Host != host { |
| 400 | s.log.Warn("refresh_host_changed", "conn", cc.id) |
| 401 | cc.close(closeRevoked) |
| 402 | s.routes.killStreams(host, closeRevoked, true) |
| 403 | return nil |
| 404 | } |
| 405 | cc.setStatus(resp) |
| 406 | s.routes.setGated(cc, resp.Status == statusGated) |
| 407 | if resp.Status == statusPaused { |
| 408 | s.routes.killStreams(host, closePaused, true) |
| 409 | } |
| 410 | if line := statementLine(resp.Access); resp.Access != "" && line != "" { |
| 411 | if err := cc.write(line); err != nil { |
| 412 | cc.close("write_error") |
| 413 | return nil |
| 414 | } |
| 415 | } |
| 416 | s.log.Info("control_refreshed", "conn", cc.id, "status", resp.Status) |
| 417 | return nil |
| 418 | } |
| 419 | |
| 420 | // sweepLoop forgets, once a minute, what the tables of sources, hints and gated days hold that nobody needs any more, so an idle node's memory shrinks. |
| 421 | func (s *Server) sweepLoop(ctx context.Context) { |
| 422 | for s.clk.sleep(ctx, time.Minute) { |
| 423 | s.ips.Sweep() |
| 424 | s.hints.Sweep() |
| 425 | s.gates.Sweep() |
| 426 | } |
| 427 | } |