| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "errors" |
| 7 | "github.com/carlosframework/platform/internal/hostwrite" |
| 8 | "log" |
| 9 | "net/http" |
| 10 | "net/http/httputil" |
| 11 | "os" |
| 12 | "path/filepath" |
| 13 | "strings" |
| 14 | "sync" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/carlosframework/platform/internal/activator" |
| 18 | "github.com/carlosframework/platform/internal/edge" |
| 19 | "github.com/carlosframework/platform/internal/provision" |
| 20 | "github.com/carlosframework/platform/internal/registry" |
| 21 | "github.com/carlosframework/platform/internal/release" |
| 22 | "github.com/carlosframework/platform/internal/schedule" |
| 23 | "github.com/carlosframework/platform/internal/version" |
| 24 | ) |
| 25 | |
| 26 | // Scheduled work, host side (spec §6.2, §6.3, §4.3). |
| 27 | // |
| 28 | // Two halves, one set of dependencies: |
| 29 | // |
| 30 | // - the CONVERGER turns apps/<acct>/<app>/schedules.json into timer rows in |
| 31 | // this box's registry. It runs on every serial move (so a `carlos schedule |
| 32 | // set` lands in seconds, like every other console-mediated write), hourly |
| 33 | // underneath that (the floor a missed bump falls back to), and once at |
| 34 | // agent start (a box that booted after the bump must not wait an hour to |
| 35 | // learn its schedules). |
| 36 | // - the RUNNER fires the rows that are due, every 15s, and reports what |
| 37 | // happened back to the bucket so `carlos schedule status` can answer |
| 38 | // without a box command. |
| 39 | // |
| 40 | // Both are built ONCE, in the agent's startup, and shared: the converger's |
| 41 | // etag memory is what keeps an idle box at one GET per app per pass and makes |
| 42 | // a run signal exactly-once, so a second instance in another loop would |
| 43 | // re-seed that memory and either replay a run or swallow one. |
| 44 | // |
| 45 | // The two halves have different preconditions. The converger needs the deploy |
| 46 | // store (schedules live in the bucket). The runner does not: once-timers are |
| 47 | // registered by instances on their own control socket (spec §7.2) straight |
| 48 | // into this box's registry, so a box with an armed activator and no store |
| 49 | // still has rows to fire — conv is nil there, and nothing is reported. |
| 50 | type scheduleDeps struct { |
| 51 | conv *schedule.Converger // nil when the box has no deploy store |
| 52 | runner *schedule.Runner |
| 53 | } |
| 54 | |
| 55 | // newScheduleDeps assembles the converger and the runner around this agent's |
| 56 | // registry, deploy store and activator. |
| 57 | // |
| 58 | // store may be nil (no deploy store on this box): no converger is built and |
| 59 | // the runner reports nothing, but it still fires the once rows instances |
| 60 | // register. svc may be nil (no hibernation armed on this box): delivery then |
| 61 | // proxies straight to the instance's socket, which is exactly right — an |
| 62 | // instance that is not activator-managed is either already running or not |
| 63 | // this leg's problem. sink may be nil (no log store): the runner simply emits |
| 64 | // no member events. Callers must pass a genuinely nil interface for sink, not |
| 65 | // a nil *applog.Sink — see the trap agent.go documents where logSink is |
| 66 | // threaded. |
| 67 | func newScheduleDeps(reg *registry.Registry, store release.Store, svc *activator.Service, sink activator.EventSink, dd string) *scheduleDeps { |
| 68 | box, err := os.Hostname() |
| 69 | if err != nil || box == "" { |
| 70 | box = "unknown" |
| 71 | } |
| 72 | var event func(rt registry.Route, level, msg string) |
| 73 | if sink != nil { |
| 74 | event = func(rt registry.Route, level, msg string) { |
| 75 | sink.Event(release.RouteAccount(rt), rt.App, rt.Host, level, msg) |
| 76 | } |
| 77 | } |
| 78 | // Zero when no activator is armed on this box: nothing hibernates, so no |
| 79 | // schedule can hold anything awake, and holdsAwake reports false for all |
| 80 | // of them. |
| 81 | var idle time.Duration |
| 82 | if svc != nil { |
| 83 | idle = svc.IdleWindow() |
| 84 | } |
| 85 | sd := &scheduleDeps{ |
| 86 | runner: &schedule.Runner{ |
| 87 | Reg: reg, |
| 88 | Deliver: &activatorDeliverer{svc: svc}, |
| 89 | Token: scheduleToken(dd, svc, reg), |
| 90 | Skip: scheduleSkip, |
| 91 | Event: event, |
| 92 | Box: box, |
| 93 | Build: version.Build, |
| 94 | Idle: idle, |
| 95 | }, |
| 96 | } |
| 97 | if store == nil { |
| 98 | return sd |
| 99 | } |
| 100 | // The reporter is shared by both halves on purpose: a converge that adds a |
| 101 | // row and a tick that fires it write the SAME status object, so whichever |
| 102 | // ran last is what a member reads. Its own timeout, not the caller's — a |
| 103 | // slow bucket must not hold a converge pass or a tick goroutine open. |
| 104 | report := func(rt registry.Route, st schedule.Status) { |
| 105 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 106 | defer cancel() |
| 107 | if err := schedule.WriteStatus(ctx, store, release.RouteAccount(rt), rt.App, st); err != nil { |
| 108 | log.Printf("schedule: report %s: %v", rt.Host, err) |
| 109 | } |
| 110 | } |
| 111 | sd.conv = schedule.NewConverger(reg, store, box, report) |
| 112 | sd.conv.Idle = idle |
| 113 | sd.runner.Report = report |
| 114 | return sd |
| 115 | } |
| 116 | |
| 117 | // armSchedule is the agent's arming switch for scheduled work, lifted out of |
| 118 | // agent.go so the decision itself can be tested rather than only its parts. |
| 119 | // |
| 120 | // storeErr is openDeployStore's error, and the three arms are not |
| 121 | // interchangeable: |
| 122 | // |
| 123 | // - a deploy store: both halves. Declarations converge into rows and every |
| 124 | // fire reports back to the bucket. |
| 125 | // - NO store but an armed activator: the RUNNER ALONE, no converger, no |
| 126 | // reports. This is the arm a review found missing (I6). A box that mounts |
| 127 | // the control socket's schedule verbs accepts once-timers from its own |
| 128 | // instances (spec §7.2) straight into the registry — and a box that |
| 129 | // accepts a timer must fire it. Gating the runner on the store would have |
| 130 | // left those rows sitting in the table forever. |
| 131 | // - neither: nothing at all, since nothing could write a row or fire one. |
| 132 | // |
| 133 | // It returns nil for that last arm, which every caller (and scheduleLegFor, |
| 134 | // and converge) already tolerates. |
| 135 | func armSchedule(reg *registry.Registry, store release.Store, storeErr error, svc *activator.Service, sink activator.EventSink, dd string) *scheduleDeps { |
| 136 | switch { |
| 137 | case storeErr == nil: |
| 138 | log.Printf("agent: scheduled work armed (converge on serial + hourly; runner every 15s)") |
| 139 | return newScheduleDeps(reg, store, svc, sink, dd) |
| 140 | case svc != nil: |
| 141 | log.Printf("agent: schedules from the bucket are off (%v); once-timers registered by instances still run (runner every 15s)", storeErr) |
| 142 | return newScheduleDeps(reg, nil, svc, sink, dd) |
| 143 | default: |
| 144 | log.Printf("agent: scheduled work disabled: %v", storeErr) |
| 145 | return nil |
| 146 | } |
| 147 | } |
| 148 | |
| 149 | // scheduleSkip is the runner's Skip (harness Run 9, defect 8): a host whose |
| 150 | // row this box holds, and a host whose offline move's record names another |
| 151 | // region. After open the source keeps its row, held, until finishing, and |
| 152 | // open has lifted the fence that used to refuse the token; so the source's |
| 153 | // tick minted a fresh <host>.env (an admin token) inside the instance |
| 154 | // directory the move had just parked, a stray secret for a host served |
| 155 | // elsewhere, and recorded a delivery the activator refuses for a held row |
| 156 | // anyway (ServeScheduled). The token is the only instance file the runner |
| 157 | // writes; delivery wakes nothing for a held row, and the converger writes |
| 158 | // registry rows and the bucket status only. Any hold counts, not only an |
| 159 | // offline move's: no held row is ever served a tick, and a hold whose move |
| 160 | // has just ended is still a hold until reconcile removes or releases the |
| 161 | // row. |
| 162 | func scheduleSkip(rt registry.Route) bool { |
| 163 | if registry.IsHold(rt.Status) { |
| 164 | return true |
| 165 | } |
| 166 | if v := armedOfflineView(); v != nil { |
| 167 | return v.elsewhere(rt.Host) |
| 168 | } |
| 169 | return false |
| 170 | } |
| 171 | |
| 172 | // scheduleToken is the runner's Token: ensure-then-read on the activator's |
| 173 | // own <dataDir>/<host>.env (spec §6.4). |
| 174 | // |
| 175 | // Ensure, not just read, because the token is otherwise minted only when the |
| 176 | // activator spawns the instance — and the tick is what would spawn it. An |
| 177 | // instance asleep since before this build rolled has no token; a read-only |
| 178 | // Token would record "no instance token" five times and give up, and the |
| 179 | // wake that mints it would never come. EnsureAdminToken is the same call |
| 180 | // Start and Seed make, serialized with them, so a tick and a visitor racing |
| 181 | // on a fresh host mint ONE token. |
| 182 | // |
| 183 | // The error is redacted: it reaches the row's last_status and from there the |
| 184 | // bucket status object and `carlos schedule ls`, and a member must not learn |
| 185 | // the box's filesystem layout from it. |
| 186 | func scheduleToken(dd string, svc *activator.Service, reg *registry.Registry) func(host string) (string, error) { |
| 187 | paths := envPathFor(dd, reg) |
| 188 | read := schedule.TokenFromEnvPaths(paths) |
| 189 | prepare := func(host string) (string, error) { |
| 190 | if err := activator.EnsureAdminToken(paths(host)); err != nil { |
| 191 | return "", errors.New(activator.RedactPaths(err.Error())) |
| 192 | } |
| 193 | tok, err := read(host) |
| 194 | if err != nil { |
| 195 | return "", errors.New(activator.RedactPaths(err.Error())) |
| 196 | } |
| 197 | return tok, nil |
| 198 | } |
| 199 | return func(host string) (string, error) { |
| 200 | if svc == nil { |
| 201 | return prepare(host) |
| 202 | } |
| 203 | return svc.PrepareScheduledToken(host, func() (string, error) { |
| 204 | return prepare(host) |
| 205 | }) |
| 206 | } |
| 207 | } |
| 208 | |
| 209 | // envPathFor is scheduleToken's route to an instance's own env file: flat by |
| 210 | // default, and the nested layout's own <host>.env (design §7) when this |
| 211 | // box's registry classifies the host as a nested row. It is the SAME |
| 212 | // classification the exec backend used to decide where it spawned the |
| 213 | // instance in the first place (activator.Classify), so the runner and the |
| 214 | // token-ensure step ask the file the backend actually wrote rather than |
| 215 | // guessing the flat path for every host. reg nil (a test with no registry |
| 216 | // in hand) is every host flat, byte-identical to before this existed. |
| 217 | func envPathFor(dd string, reg *registry.Registry) func(host string) string { |
| 218 | flat := func(host string) string { return filepath.Join(dd, host+".env") } |
| 219 | if reg == nil { |
| 220 | return flat |
| 221 | } |
| 222 | return func(host string) string { |
| 223 | rt, err := reg.Get(host) |
| 224 | if err != nil || rt == nil { |
| 225 | return flat(host) |
| 226 | } |
| 227 | if layout, _ := activator.Classify(*rt, activator.LayoutInputs{DataDir: dd}); layout == activator.LayoutNested { |
| 228 | return filepath.Join(provision.InstanceDir(dd, host), host+".env") |
| 229 | } |
| 230 | return flat(host) |
| 231 | } |
| 232 | } |
| 233 | |
| 234 | // convergePassTimeout bounds ONE converge pass. The agent's three call sites |
| 235 | // (the seed and the serial leg in fastConvergeLoop, the hourly floor in |
| 236 | // adoptLoop) all hand this context.Background() — an unbounded context — and |
| 237 | // a pass is store-bound: a bucket that accepts a connection and then never |
| 238 | // answers would park the pass, and with it every caller queued on |
| 239 | // Converger's own mutex, for as long as the process lives. Generous on |
| 240 | // purpose: a pass reads one object per app plus the run signal, so minutes |
| 241 | // is far past any healthy bucket and still finite. The floors underneath |
| 242 | // (hourly, and the next serial move) are what recover a pass cut short. |
| 243 | // |
| 244 | // It bounds QUEUE PLUS WORK, not work alone: Pass takes the converger's |
| 245 | // mutex first, so a caller behind a parked pass can spend the budget waiting |
| 246 | // and then run with an expired context. Harmless — every app read fails, |
| 247 | // convergeApp is die-free and leaves its rows alone, no etag is poisoned — |
| 248 | // and it is the property that matters here: the leg returns, whatever the |
| 249 | // bucket is doing. |
| 250 | const convergePassTimeout = 5 * time.Minute |
| 251 | |
| 252 | // converge runs one converger pass, or nothing on a box with no deploy |
| 253 | // store. The agent's three call sites go through this so none of them can |
| 254 | // reach into a nil conv — and so that all three get the timeout above |
| 255 | // rather than each remembering to apply it. |
| 256 | func (sd *scheduleDeps) converge(ctx context.Context) error { |
| 257 | if sd == nil || sd.conv == nil { |
| 258 | return nil |
| 259 | } |
| 260 | ctx, cancel := context.WithTimeout(ctx, convergePassTimeout) |
| 261 | defer cancel() |
| 262 | return sd.conv.Pass(ctx) |
| 263 | } |
| 264 | |
| 265 | // scheduleLegFor is fastDeps.scheduleLeg for sd — a genuinely nil func when |
| 266 | // there is nothing to converge (no scheduled work armed, or no deploy store |
| 267 | // to read declarations from), so the tick skips the leg rather than calling |
| 268 | // a closure over a nil converger. |
| 269 | func scheduleLegFor(sd *scheduleDeps) func(ctx context.Context) { |
| 270 | if sd == nil || sd.conv == nil { |
| 271 | return nil |
| 272 | } |
| 273 | return func(ctx context.Context) { |
| 274 | if err := sd.converge(ctx); err != nil { |
| 275 | log.Printf("agent: schedule converge: %v", err) |
| 276 | } |
| 277 | } |
| 278 | } |
| 279 | |
| 280 | // activatorDeliverer delivers one tick the way a member request reaches an |
| 281 | // instance from inside this process — through the activator, then the |
| 282 | // instance's unix socket — and never through the public edge (spec §6.3). |
| 283 | // Two properties come from that and only that: |
| 284 | // |
| 285 | // - a hibernating instance WAKES for its tick, exactly as it would for a |
| 286 | // visitor, and is held awake for the tick's duration because the |
| 287 | // activator's in-flight count holds it open — but the tick earns no |
| 288 | // activity credit, so it does not by itself keep the instance awake |
| 289 | // afterwards. Separately, the sweeper's schedule guard (activator.go, |
| 290 | // sweepOnce) holds a host resident whenever its next tick is due sooner |
| 291 | // than the idle window, so a schedule firing more often than the idle |
| 292 | // window stays resident by that guard's design rather than thrashing |
| 293 | // through a wake/sleep pair on every tick. A schedule looser than the |
| 294 | // idle window is not covered by the guard and does sleep between ticks, |
| 295 | // waking again for the next one. |
| 296 | // - the tick never leaves the box: no DNS, no TLS, no edge auth, nothing an |
| 297 | // outsider could aim at the same path. |
| 298 | type activatorDeliverer struct { |
| 299 | svc *activator.Service |
| 300 | |
| 301 | // proxies caches one unix proxy — and therefore one http.Transport, and |
| 302 | // therefore one connection pool — per socket path. A fresh Transport per |
| 303 | // delivery meant a fresh connection for every tick and a pool thrown |
| 304 | // away unread; a schedule that fires every minute paid a dial (and the |
| 305 | // runtime a reader goroutine) each time for nothing. |
| 306 | // Keyed by socket path rather than host because the path is what the |
| 307 | // Transport actually dials: a route re-pointed at a new socket gets a new |
| 308 | // entry rather than a pool aimed at the old file. |
| 309 | // |
| 310 | // Bounded three ways, in the order they normally fire: |
| 311 | // |
| 312 | // - the runner's daily prune leg hands this the sockets still routed |
| 313 | // here (RetainSockets) and everything else goes. That is the one that |
| 314 | // reclaims a RETIRED route: it stops receiving deliveries, so it can |
| 315 | // never error its own way out. |
| 316 | // - an entry is dropped when its delivery comes back as an error — a |
| 317 | // refused wake or an expired deadline. (A socket that will not dial |
| 318 | // is NOT one of these: ReverseProxy answers that itself with a 502 |
| 319 | // and no error. Nothing is pooled behind a failed dial anyway.) |
| 320 | // - the whole map is dropped if it ever passes maxProxyCache, a safety |
| 321 | // net rather than a working mechanism. |
| 322 | // |
| 323 | // Every one of those closes the dropped transports' idle connections |
| 324 | // rather than waiting out edge.UnixProxy's 90s IdleConnTimeout. |
| 325 | mu sync.Mutex |
| 326 | proxies map[string]http.Handler |
| 327 | |
| 328 | // genPools is the delivery leg's own per-generation-socket pools, used |
| 329 | // instead of proxies for an activator-managed instance once generations |
| 330 | // are on (zero-downtime spec §7.1): a tick reaches the CURRENT |
| 331 | // generation, asked per connection. Made on first use and retired |
| 332 | // through the activator's hook, so a retired generation's pool closes |
| 333 | // with it rather than waiting for the daily prune. Nil with generations |
| 334 | // off, and nothing below reads it then. |
| 335 | genOnce sync.Once |
| 336 | genPools *edge.GenPools |
| 337 | } |
| 338 | |
| 339 | // managedByGenerations reports whether rt's ticks go through the current |
| 340 | // generation: generations on, and an activator-managed instance. |
| 341 | func (d *activatorDeliverer) managedByGenerations(rt registry.Route) bool { |
| 342 | return d.svc != nil && d.svc.GenerationsOn() && rt.Kind == "instance" && rt.Hibernate |
| 343 | } |
| 344 | |
| 345 | // genProxyFor is the AppSocket proxy for host over the leg's own pools. |
| 346 | func (d *activatorDeliverer) genProxyFor(host string) http.Handler { |
| 347 | d.genOnce.Do(func() { |
| 348 | d.genPools = edge.NewGenPools() |
| 349 | d.svc.OnGenRetired(d.genPools.Retire) |
| 350 | }) |
| 351 | return edge.AppSocketProxy(host, d.svc, d.genPools) |
| 352 | } |
| 353 | |
| 354 | // maxProxyCache is that ceiling. Deliberately well above the number of |
| 355 | // exec instances a box runs (the 1000-per-host timer cap is per host, not |
| 356 | // per box) so the reset is a safety net, never a working mechanism. |
| 357 | const maxProxyCache = 512 |
| 358 | |
| 359 | // proxyFor is the cached unix proxy for socket. |
| 360 | func (d *activatorDeliverer) proxyFor(socket string) http.Handler { |
| 361 | d.mu.Lock() |
| 362 | defer d.mu.Unlock() |
| 363 | if p, ok := d.proxies[socket]; ok { |
| 364 | return p |
| 365 | } |
| 366 | if d.proxies == nil { |
| 367 | d.proxies = map[string]http.Handler{} |
| 368 | } else if len(d.proxies) >= maxProxyCache { |
| 369 | for _, p := range d.proxies { |
| 370 | closeIdle(p) |
| 371 | } |
| 372 | clear(d.proxies) |
| 373 | } |
| 374 | p := edge.UnixProxy(socket) |
| 375 | d.proxies[socket] = p |
| 376 | return p |
| 377 | } |
| 378 | |
| 379 | // dropProxy forgets socket's proxy and closes the connections it was holding. |
| 380 | func (d *activatorDeliverer) dropProxy(socket string) { |
| 381 | d.mu.Lock() |
| 382 | defer d.mu.Unlock() |
| 383 | if p, ok := d.proxies[socket]; ok { |
| 384 | closeIdle(p) |
| 385 | delete(d.proxies, socket) |
| 386 | } |
| 387 | } |
| 388 | |
| 389 | // RetainSockets keeps only the sockets that are still routed on this box and |
| 390 | // drops the rest — schedule.SocketRetainer, called by the runner's prune leg |
| 391 | // with the host list it has already read. Without it a retired route's entry |
| 392 | // would sit here until the maxProxyCache reset, since a route nothing |
| 393 | // delivers to can never error its way out. |
| 394 | func (d *activatorDeliverer) RetainSockets(live map[string]bool) { |
| 395 | d.mu.Lock() |
| 396 | defer d.mu.Unlock() |
| 397 | for socket, p := range d.proxies { |
| 398 | if !live[socket] { |
| 399 | closeIdle(p) |
| 400 | delete(d.proxies, socket) |
| 401 | } |
| 402 | } |
| 403 | } |
| 404 | |
| 405 | // closeIdle returns a dropped proxy's pooled connections now rather than at |
| 406 | // the end of edge.UnixProxy's 90s IdleConnTimeout. Best-effort by design: |
| 407 | // both assertions are guarded, so a change to the shape edge builds costs |
| 408 | // the eager close and nothing else — the idle timeout is still the backstop. |
| 409 | // (schedule_wake_test.go pins the shape, so such a change fails a test |
| 410 | // rather than passing quietly.) |
| 411 | func closeIdle(h http.Handler) { |
| 412 | if rp, ok := h.(*httputil.ReverseProxy); ok { |
| 413 | if tr, ok := rp.Transport.(*http.Transport); ok { |
| 414 | tr.CloseIdleConnections() |
| 415 | } |
| 416 | } |
| 417 | } |
| 418 | |
| 419 | func (d *activatorDeliverer) Deliver(ctx context.Context, rt registry.Route, req *http.Request) (int, error) { |
| 420 | return d.deliverSelected(ctx, rt, req, nil) |
| 421 | } |
| 422 | func (d *activatorDeliverer) DeliverClaim(ctx context.Context, rt registry.Route, req *http.Request, admit func(context.Context) (context.Context, func(), error)) (int, error) { |
| 423 | return d.deliverSelected(ctx, rt, req, admit) |
| 424 | } |
| 425 | func (d *activatorDeliverer) deliverSelected(ctx context.Context, rt registry.Route, req *http.Request, admit func(context.Context) (context.Context, func(), error)) (int, error) { |
| 426 | if rt.Socket == "" { |
| 427 | // An addr-backed (or tunnel-backed) route has no socket to dial from |
| 428 | // here. Converge only ever writes rows for exec-backed instances, so |
| 429 | // this is a row that outlived its route's shape — an error, not a dial |
| 430 | // to an empty path (which would be a connect-refused every 15s). |
| 431 | return 0, errors.New("no socket for route " + rt.Host) |
| 432 | } |
| 433 | // Generations on: the proxy asks the activator for the current |
| 434 | // generation's socket per connection (spec §7.1). Off: the cached proxy |
| 435 | // on rt.Socket, exactly as before. |
| 436 | gen := d.managedByGenerations(rt) |
| 437 | var proxy http.Handler |
| 438 | if gen { |
| 439 | proxy = d.genProxyFor(rt.Host) |
| 440 | } else { |
| 441 | proxy = d.proxyFor(rt.Socket) |
| 442 | } |
| 443 | var admissionErr error |
| 444 | if admit != nil { |
| 445 | next := proxy |
| 446 | proxy = http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 447 | call, finish, err := admit(r.Context()) |
| 448 | if err != nil { |
| 449 | admissionErr = errors.Join(hostwrite.ErrUnavailable, err) |
| 450 | http.Error(w, "instance timer authority changed", http.StatusServiceUnavailable) |
| 451 | return |
| 452 | } |
| 453 | defer finish() |
| 454 | next.ServeHTTP(w, r.WithContext(call)) |
| 455 | }) |
| 456 | } |
| 457 | h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 458 | // ServeScheduled, not ServeHTTP: a tick wakes the instance and is |
| 459 | // held in flight for its duration, but must not reset the idle |
| 460 | // clock. Reports false for a route it does not manage (not an |
| 461 | // instance, or not hibernate-flagged); that route is proxied |
| 462 | // straight, the same call shape the edge itself uses. |
| 463 | if d.svc == nil || !d.svc.ServeScheduled(w, r, rt, proxy) { |
| 464 | proxy.ServeHTTP(w, r) |
| 465 | } |
| 466 | }) |
| 467 | status, err := deliverVia(ctx, h, rt, req) |
| 468 | if admissionErr != nil { |
| 469 | return 0, admissionErr |
| 470 | } |
| 471 | if err != nil { |
| 472 | // Anything that did not come back as an HTTP status — a wake the |
| 473 | // activator refused, a delivery that ran past its deadline — retires |
| 474 | // this socket's pool; the next tick dials fresh. Not the case a |
| 475 | // reader expects here: a socket that will not dial is ReverseProxy's |
| 476 | // own 502 with no error, and it has nothing pooled to retire anyway. |
| 477 | // Retired ROUTES are reclaimed by RetainSockets, not from here. A |
| 478 | // generation's pool is retired by the activator's hook instead. |
| 479 | if !gen { |
| 480 | d.dropProxy(rt.Socket) |
| 481 | } |
| 482 | } |
| 483 | if err != nil && ctx.Err() == nil { |
| 484 | // Redacted for the same reason as the token error: this string is |
| 485 | // the row's last_status. The context's own error is left alone — |
| 486 | // the runner matches it with errors.Is to tell a timeout apart. |
| 487 | err = errors.New(activator.RedactPaths(err.Error())) |
| 488 | } |
| 489 | return status, err |
| 490 | } |
| 491 | |
| 492 | // deliverVia is the pure core: run the handler against a recorder and turn |
| 493 | // what it wrote into the runner's (status, error) pair. |
| 494 | // |
| 495 | // The mapping is the whole point. The activator answers its own refusals with |
| 496 | // a 503 whose body names the instance ("instance is waking", "instance failed |
| 497 | // to wake", "instance is waking elsewhere"), which is indistinguishable at the |
| 498 | // status line from an app that answered 503 itself. Both retry — but the |
| 499 | // member-facing status must say WHICH, so a wake refusal comes back as an |
| 500 | // error (classified "503: instance is waking…") and an app's own 503 comes |
| 501 | // back as the status 503. Anything else is returned as the app said it. |
| 502 | func deliverVia(ctx context.Context, h http.Handler, rt registry.Route, req *http.Request) (int, error) { |
| 503 | rec := &statusRecorder{} |
| 504 | h.ServeHTTP(rec, req) |
| 505 | // Checked AFTER the handler returns, and before the status: a tick that |
| 506 | // ran past its deadline gets a 502 written under it by the proxy on the |
| 507 | // way out, and reporting that 502 would retry a job that may well have |
| 508 | // completed. The runner maps DeadlineExceeded to ErrTimeout, the one |
| 509 | // outcome that does not retry. |
| 510 | if err := ctx.Err(); err != nil { |
| 511 | return 0, err |
| 512 | } |
| 513 | line := rec.firstLine() |
| 514 | if rec.status == http.StatusServiceUnavailable && strings.HasPrefix(line, "instance") { |
| 515 | return 0, errors.New(line) |
| 516 | } |
| 517 | if rec.status == 0 { |
| 518 | // The handler returned without writing anything and the context is |
| 519 | // fine — nothing an instance can legitimately do (even a failed dial |
| 520 | // is a 502). Reporting status 0 would classify as "app refused" and |
| 521 | // never retry; this says what actually happened and retries. |
| 522 | return 0, errors.New("no response from " + rt.Host) |
| 523 | } |
| 524 | return rec.status, nil |
| 525 | } |
| 526 | |
| 527 | // firstLineMax bounds what the recorder keeps of a body. A tick's response |
| 528 | // body is never delivered anywhere — only the first line of a wake refusal is |
| 529 | // ever read — so a chatty app cannot make this leg hold its output in memory. |
| 530 | const firstLineMax = 256 |
| 531 | |
| 532 | // statusRecorder is the ResponseWriter the activator and the proxy write |
| 533 | // through: it keeps the status and the first line of the body and discards |
| 534 | // everything else. Not httptest.ResponseRecorder — that one buffers the whole |
| 535 | // body, and this runs in the agent on every tick. |
| 536 | type statusRecorder struct { |
| 537 | header http.Header |
| 538 | status int |
| 539 | line []byte |
| 540 | lineEnd bool // newline seen or cap reached — later writes are discarded |
| 541 | } |
| 542 | |
| 543 | func (r *statusRecorder) Header() http.Header { |
| 544 | if r.header == nil { |
| 545 | r.header = http.Header{} |
| 546 | } |
| 547 | return r.header |
| 548 | } |
| 549 | |
| 550 | func (r *statusRecorder) WriteHeader(code int) { |
| 551 | if r.status == 0 { |
| 552 | r.status = code |
| 553 | } |
| 554 | } |
| 555 | |
| 556 | func (r *statusRecorder) Write(b []byte) (int, error) { |
| 557 | r.WriteHeader(http.StatusOK) // implicit 200, as on a real writer |
| 558 | if !r.lineEnd { |
| 559 | if i := bytes.IndexByte(b, '\n'); i >= 0 { |
| 560 | r.line = append(r.line, b[:i]...) |
| 561 | r.lineEnd = true |
| 562 | } else { |
| 563 | r.line = append(r.line, b...) |
| 564 | } |
| 565 | if len(r.line) >= firstLineMax { |
| 566 | r.line = r.line[:firstLineMax] |
| 567 | r.lineEnd = true |
| 568 | } |
| 569 | } |
| 570 | return len(b), nil |
| 571 | } |
| 572 | |
| 573 | // firstLine is the body's first line, trimmed — what the wake-refusal test |
| 574 | // above matches on. |
| 575 | func (r *statusRecorder) firstLine() string { return strings.TrimSpace(string(r.line)) } |
| 576 | |
amadan