| 1 | package edge |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "net" |
| 9 | "net/http" |
| 10 | "net/http/httputil" |
| 11 | "sync" |
| 12 | "syscall" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/unixdial" |
| 16 | ) |
| 17 | |
| 18 | // genproxy.go is the edge half of generations (zero-downtime spec §7.2): a |
| 19 | // reverse proxy for an activator-managed route whose app socket moves from |
| 20 | // one generation to the next. Every request asks the activator for the |
| 21 | // current generation and uses that generation's pool, whose connections all |
| 22 | // reach that generation's socket (dialGen), so the response is attributed to |
| 23 | // the generation that served it. A connection to generation n is never |
| 24 | // reused for n+1, and a generation's idle connections are closed when it |
| 25 | // stops being current. |
| 26 | // |
| 27 | // Used only when the activator's GenerationsOn is true. With it false the |
| 28 | // edge builds unixProxy(t.Socket) exactly as before. |
| 29 | |
| 30 | // errNoGeneration is a dial's answer while the route has no current |
| 31 | // generation (between a stop and the next wake). It wraps ENOENT on purpose: |
| 32 | // to dialUnixRetrying it is the same "socket briefly absent" gap a restart |
| 33 | // always had, retried inside the same window, and each retry asks again. |
| 34 | var errNoGeneration = fmt.Errorf("edge: instance has no current generation: %w", syscall.ENOENT) |
| 35 | |
| 36 | // dialUnixRetrying is dialUnixWithRetry with the socket resolved per attempt: |
| 37 | // resolve is asked before the first dial and again before every retry, so a |
| 38 | // dial that races a switch lands on the generation that is current by then. |
| 39 | func dialUnixRetrying(ctx context.Context, d *net.Dialer, resolve func() (string, bool)) (net.Conn, error) { |
| 40 | const window = 13 * time.Second |
| 41 | deadline := time.Now().Add(window) |
| 42 | for { |
| 43 | var ( |
| 44 | conn net.Conn |
| 45 | err error |
| 46 | ) |
| 47 | if sock, ok := resolve(); ok { |
| 48 | conn, err = unixdial.Checked(ctx, d, sock) |
| 49 | } else { |
| 50 | err = errNoGeneration |
| 51 | } |
| 52 | if err == nil { |
| 53 | return conn, nil |
| 54 | } |
| 55 | if ctx.Err() != nil || !isRetryableDialErr(err) || time.Now().After(deadline) { |
| 56 | return nil, err |
| 57 | } |
| 58 | select { |
| 59 | case <-ctx.Done(): |
| 60 | return nil, ctx.Err() |
| 61 | case <-time.After(100 * time.Millisecond): |
| 62 | } |
| 63 | } |
| 64 | } |
| 65 | |
| 66 | // genMovedError is a generation pool's dial giving up: its own socket is |
| 67 | // gone and another generation is current. Nothing was sent, so the request |
| 68 | // can go again on the current generation's pool. |
| 69 | type genMovedError struct{ to GenInfo } |
| 70 | |
| 71 | func (e *genMovedError) Error() string { |
| 72 | return fmt.Sprintf("edge: generation moved to %d while dialling the one before it", e.to.N) |
| 73 | } |
| 74 | |
| 75 | // dialGen is a generation pool's dial: own's socket and nothing else, and |
| 76 | // a connection proven to be own's (a shared foreign path is checked |
| 77 | // against the current generation once connected), so every connection in |
| 78 | // the pool is own's and a response through it is attributed to own exactly |
| 79 | // (spec §7.2). The retiring generation still |
| 80 | // accepts until its SIGTERM, which comes after the switch, so a request |
| 81 | // that chose own's pool an instant before the switch is served there. A |
| 82 | // retryable failure is retried inside the usual window while own is still |
| 83 | // current or no generation is; once another one is current, own is gone for |
| 84 | // good and the dial returns genMovedError. |
| 85 | func dialGen(ctx context.Context, d *net.Dialer, src GenSource, host string, own GenInfo) (net.Conn, error) { |
| 86 | const window = 13 * time.Second |
| 87 | deadline := time.Now().Add(window) |
| 88 | for { |
| 89 | conn, err := unixdial.Checked(ctx, d, own.Socket) |
| 90 | if err == nil { |
| 91 | // Spec §7.2 (rev 17): if the generation stops being current |
| 92 | // before the dial completes, the request is retried against the |
| 93 | // new current generation. So every new connection is checked |
| 94 | // once connected: own must still be current. This also covers a |
| 95 | // foreign row, whose generations share one canonical path, so a |
| 96 | // connection on it could have reached the replacement. A |
| 97 | // connection already made (an idle one in own's pool) is not |
| 98 | // re-checked: the request that chose own's pool sampled own. |
| 99 | cur, ok := src.ServingGen(host) |
| 100 | if ok && cur.N == own.N { |
| 101 | return conn, nil |
| 102 | } |
| 103 | conn.Close() |
| 104 | if ok { |
| 105 | return nil, &genMovedError{to: cur} |
| 106 | } |
| 107 | err = errNoGeneration |
| 108 | } |
| 109 | if ctx.Err() != nil || !isRetryableDialErr(err) || time.Now().After(deadline) { |
| 110 | return nil, err |
| 111 | } |
| 112 | if cur, ok := src.ServingGen(host); ok && (cur.N != own.N || cur.Socket != own.Socket) { |
| 113 | return nil, &genMovedError{to: cur} |
| 114 | } |
| 115 | select { |
| 116 | case <-ctx.Done(): |
| 117 | return nil, ctx.Err() |
| 118 | case <-time.After(100 * time.Millisecond): |
| 119 | } |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | // genPool is one generation's transport and the generation it reaches. |
| 124 | type genPool struct { |
| 125 | tr *http.Transport |
| 126 | gen GenInfo |
| 127 | } |
| 128 | |
| 129 | // GenPools is the per-socket-path set of transports, one idle pool per |
| 130 | // generation socket. One set per consumer (the edge has one, scheduled |
| 131 | // delivery has its own) and shared across route reloads, because the pool |
| 132 | // belongs to the generation, not to the table build that first reached it. |
| 133 | type GenPools struct { |
| 134 | mu sync.Mutex |
| 135 | m map[string]*genPool |
| 136 | } |
| 137 | |
| 138 | // NewGenPools returns an empty set. |
| 139 | func NewGenPools() *GenPools { return &GenPools{m: map[string]*genPool{}} } |
| 140 | |
| 141 | // pool is generation g's pool, made on first use, or false when g is no |
| 142 | // longer the host's current generation. Its dial reaches g's own socket only |
| 143 | // (dialGen). A foreign row reuses its canonical path for every generation |
| 144 | // (plan 1's reading), so a pool found for the path under another generation |
| 145 | // number is replaced, and its idle connections closed. |
| 146 | // |
| 147 | // g was sampled by the caller some time ago. The check against src happens |
| 148 | // under p.mu, and Retire takes the same lock, so a pool is published only for |
| 149 | // a generation that was still current at that instant: if a switch has |
| 150 | // retired g since, either Retire has already run (and this refuses to publish |
| 151 | // a pool nothing would ever remove) or it runs after the publish and removes |
| 152 | // it. The caller samples again on false. |
| 153 | // |
| 154 | // Lock order: p.mu, then whatever lock src.ServingGen takes (the activator's |
| 155 | // own). That is safe because Retire runs from OnGenRetired hooks, which the |
| 156 | // activator calls outside every lock of its own, and ServingGen never calls |
| 157 | // into this set. A GenSource whose ServingGen reached Retire would deadlock. |
| 158 | func (p *GenPools) pool(host string, g GenInfo, src GenSource) (*genPool, bool) { |
| 159 | p.mu.Lock() |
| 160 | defer p.mu.Unlock() |
| 161 | if cur, ok := src.ServingGen(host); !ok || cur.N != g.N { |
| 162 | return nil, false |
| 163 | } |
| 164 | if gp, ok := p.m[g.Socket]; ok { |
| 165 | if gp.gen.N == g.N { |
| 166 | return gp, true |
| 167 | } |
| 168 | gp.tr.CloseIdleConnections() |
| 169 | } |
| 170 | gp := &genPool{tr: genTransportFor(host, src, &g), gen: g} |
| 171 | p.m[g.Socket] = gp |
| 172 | return gp, true |
| 173 | } |
| 174 | |
| 175 | // Retire closes socket's idle connections and forgets its pool. It has the |
| 176 | // shape of GenSource.OnGenRetired's hook. A request still in flight on a |
| 177 | // retired pool finishes; its connection is not reused by anyone after. |
| 178 | func (p *GenPools) Retire(_, socket string) { |
| 179 | p.mu.Lock() |
| 180 | gp, ok := p.m[socket] |
| 181 | delete(p.m, socket) |
| 182 | p.mu.Unlock() |
| 183 | if ok { |
| 184 | gp.tr.CloseIdleConnections() |
| 185 | } |
| 186 | } |
| 187 | |
| 188 | // Len is the number of live pools. |
| 189 | func (p *GenPools) Len() int { |
| 190 | p.mu.Lock() |
| 191 | defer p.mu.Unlock() |
| 192 | return len(p.m) |
| 193 | } |
| 194 | |
| 195 | // genTransportFor builds the transport behind one pool: unixProxy's idle |
| 196 | // bounds, and a dial that reaches own's socket only (dialGen). RoundTrip |
| 197 | // always passes own: a request that finds no current generation waits for |
| 198 | // one (waitServingGen) and then uses that generation's pool. With own nil |
| 199 | // it is a one-shot transport whose dial resolves the socket per attempt. |
| 200 | func genTransportFor(host string, src GenSource, own *GenInfo) *http.Transport { |
| 201 | dialer := &net.Dialer{} |
| 202 | dial := func(ctx context.Context, _, _ string) (net.Conn, error) { |
| 203 | return dialUnixRetrying(ctx, dialer, func() (string, bool) { return src.AppSocket(host) }) |
| 204 | } |
| 205 | if own != nil { |
| 206 | g := *own |
| 207 | dial = func(ctx context.Context, _, _ string) (net.Conn, error) { |
| 208 | return dialGen(ctx, dialer, src, host, g) |
| 209 | } |
| 210 | } |
| 211 | return &http.Transport{ |
| 212 | DialContext: dial, |
| 213 | MaxIdleConns: 32, |
| 214 | IdleConnTimeout: 90 * time.Second, |
| 215 | DisableKeepAlives: own == nil, |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | // genRoundTripper picks the pool of host's current generation for each |
| 220 | // request, stamps the response with the build of the generation whose |
| 221 | // connection carried it, and records that generation on the request's |
| 222 | // outcome for the cache (spec §7.2). A pool's connections are all its own |
| 223 | // generation's (dialGen), so the pool's generation is that generation. |
| 224 | type genRoundTripper struct { |
| 225 | host string |
| 226 | src GenSource |
| 227 | pools *GenPools |
| 228 | } |
| 229 | |
| 230 | // genRetries bounds how often one request is sent again after its |
| 231 | // generation moved under it. |
| 232 | const genRetries = 2 |
| 233 | |
| 234 | // genStaleRetries bounds how often one request goes back to choosing a |
| 235 | // generation because the one it chose was no longer current when its pool was |
| 236 | // asked for. That is not a dial attempt, so it does not use genRetries, but a |
| 237 | // source that keeps switching must not hold the request forever. |
| 238 | const genStaleRetries = 3 |
| 239 | |
| 240 | func (t *genRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { |
| 241 | // The generation is chosen, and any switch retried, before a byte of the |
| 242 | // body is consumed (spec §7.2, rev 17). The transport gets the body |
| 243 | // behind heldBody: it reads it only once it is writing the request on a |
| 244 | // connection dialGen has proven to be the chosen generation's, and a |
| 245 | // Close it makes after a failed dial is held. So a dial that finds the |
| 246 | // generation moved leaves the body untouched, and the request goes again |
| 247 | // on the current generation's pool, a streaming POST with no GetBody |
| 248 | // included. |
| 249 | var hb *heldBody |
| 250 | if req.Body != nil && req.Body != http.NoBody { |
| 251 | hb = &heldBody{rc: req.Body} |
| 252 | req.Body = hb |
| 253 | } |
| 254 | for attempt, stale := 0, 0; ; { |
| 255 | g, ok := t.src.ServingGen(t.host) |
| 256 | if !ok { |
| 257 | // No current generation (between a stop and the next wake): wait |
| 258 | // for one inside the dial's retry window, then take the normal |
| 259 | // path, so the response is attributed to the generation that |
| 260 | // serves it and its connection passes dialGen's check (spec §7.2). |
| 261 | g, ok = waitServingGen(req.Context(), t.src, t.host) |
| 262 | if !ok { |
| 263 | if hb != nil { |
| 264 | hb.release() |
| 265 | } |
| 266 | return nil, errNoGeneration |
| 267 | } |
| 268 | } |
| 269 | gp, ok := t.pools.pool(t.host, g, t.src) |
| 270 | if !ok { |
| 271 | // g was retired between the choice and the pool: choose again. |
| 272 | if stale++; stale > genStaleRetries { |
| 273 | if hb != nil { |
| 274 | hb.release() |
| 275 | } |
| 276 | if cur, ok := t.src.ServingGen(t.host); ok { |
| 277 | return nil, &genMovedError{to: cur} |
| 278 | } |
| 279 | return nil, errNoGeneration |
| 280 | } |
| 281 | continue |
| 282 | } |
| 283 | resp, err := gp.tr.RoundTrip(req) |
| 284 | var moved *genMovedError |
| 285 | if errors.As(err, &moved) && attempt < genRetries && (hb == nil || !hb.wasRead()) { |
| 286 | // The generation this request chose stopped being current before |
| 287 | // its dial completed. Nothing was sent and nothing of the body |
| 288 | // was read: send it again, on the current one. |
| 289 | attempt++ |
| 290 | continue |
| 291 | } |
| 292 | if err != nil { |
| 293 | if hb != nil { |
| 294 | hb.release() |
| 295 | } |
| 296 | return nil, err |
| 297 | } |
| 298 | // X-Carlos-Version names the build of the generation that served the |
| 299 | // request, not the registry's Version (spec §7.2). Set, not Add: the |
| 300 | // platform's voice on this header is the only one. |
| 301 | if gp.gen.Build != "" { |
| 302 | resp.Header.Set("X-Carlos-Version", gp.gen.Build) |
| 303 | } |
| 304 | noteServedGen(req, t.host, gp.gen.N) |
| 305 | return resp, nil |
| 306 | } |
| 307 | } |
| 308 | |
| 309 | // waitServingGen waits, within the dial's 13 s retry window and ctx, for |
| 310 | // host to have a current generation, asking every 100 ms. |
| 311 | func waitServingGen(ctx context.Context, src GenSource, host string) (GenInfo, bool) { |
| 312 | deadline := time.Now().Add(13 * time.Second) |
| 313 | for { |
| 314 | if g, ok := src.ServingGen(host); ok { |
| 315 | return g, true |
| 316 | } |
| 317 | if time.Now().After(deadline) { |
| 318 | return GenInfo{}, false |
| 319 | } |
| 320 | select { |
| 321 | case <-ctx.Done(): |
| 322 | return GenInfo{}, false |
| 323 | case <-time.After(100 * time.Millisecond): |
| 324 | } |
| 325 | } |
| 326 | } |
| 327 | |
| 328 | // heldBody is a request body the transport cannot consume or close before |
| 329 | // the request is on a connection. The transport reads a body only while it |
| 330 | // writes the request on a dialled connection, and closes it on any error, |
| 331 | // including a dial that never connected. Until the first Read, that Close |
| 332 | // is held, so the same body can go again on another generation's pool. |
| 333 | // After it, Close passes through as usual. |
| 334 | type heldBody struct { |
| 335 | rc io.ReadCloser |
| 336 | mu sync.Mutex |
| 337 | read bool |
| 338 | } |
| 339 | |
| 340 | func (b *heldBody) Read(p []byte) (int, error) { |
| 341 | b.mu.Lock() |
| 342 | b.read = true |
| 343 | b.mu.Unlock() |
| 344 | return b.rc.Read(p) |
| 345 | } |
| 346 | |
| 347 | func (b *heldBody) Close() error { |
| 348 | b.mu.Lock() |
| 349 | read := b.read |
| 350 | b.mu.Unlock() |
| 351 | if !read { |
| 352 | return nil // a failed dial's close: the body may still be sent |
| 353 | } |
| 354 | return b.rc.Close() |
| 355 | } |
| 356 | |
| 357 | func (b *heldBody) wasRead() bool { |
| 358 | b.mu.Lock() |
| 359 | defer b.mu.Unlock() |
| 360 | return b.read |
| 361 | } |
| 362 | |
| 363 | // release closes the body for good: the request has finished with an error. |
| 364 | func (b *heldBody) release() { _ = b.rc.Close() } |
| 365 | |
| 366 | // appSocketProxy is unixProxy for an activator-managed route with |
| 367 | // generations on: the same Rewrite, Host preservation, upgrade pass-through |
| 368 | // and error handler, over genRoundTripper instead of one fixed socket. |
| 369 | func appSocketProxy(host string, src GenSource, pools *GenPools) *httputil.ReverseProxy { |
| 370 | return &httputil.ReverseProxy{ |
| 371 | Rewrite: func(pr *httputil.ProxyRequest) { |
| 372 | setForwarded(pr) |
| 373 | pr.Out.URL.Scheme = "http" |
| 374 | pr.Out.URL.Host = "carlos" // placeholder; the unix dialer ignores it |
| 375 | pr.Out.Host = pr.In.Host // backend sees the real hostname |
| 376 | scrubInboundPurgeHeader(pr) |
| 377 | }, |
| 378 | Transport: &genRoundTripper{host: host, src: src, pools: pools}, |
| 379 | ErrorHandler: defaultProxyError(proxyCause), |
| 380 | } |
| 381 | } |
| 382 | |
| 383 | // AppSocketProxy is appSocketProxy for callers outside this package: the |
| 384 | // scheduled-work deliverer in internal/host, which must reach the current |
| 385 | // generation the same way the edge does (spec §7.1) without the edge's |
| 386 | // cache, compression or usage layers. |
| 387 | func AppSocketProxy(host string, src GenSource, pools *GenPools) http.Handler { |
| 388 | return appSocketProxy(host, src, pools) |
| 389 | } |
| 390 | |
amadan