| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "sync/atomic" |
| 6 | |
| 7 | "github.com/carlosframework/platform/internal/edge" |
| 8 | "github.com/carlosframework/platform/internal/registry" |
| 9 | "github.com/carlosframework/platform/internal/release" |
| 10 | ) |
| 11 | |
| 12 | // The Service's generation surface (zero-downtime spec §7.1). AppSocket is |
| 13 | // for traffic; the handle (Gen) is for whoever started the generation. |
| 14 | var _ edge.GenSource = (*Service)(nil) |
| 15 | |
| 16 | // GenerationsOn is the dormant switch as this Service read it at New. |
| 17 | func (a *Service) GenerationsOn() bool { return a.gens } |
| 18 | |
| 19 | // startGen spawns the next generation of rt's instance: the process-wide |
| 20 | // counter's next number, and the backend's generation spawn when it has |
| 21 | // one. A backend without one is spawned with Start and gets the canonical |
| 22 | // handle. |
| 23 | func (a *Service) startGen(ctx context.Context, rt registry.Route) (Proc, Gen, error) { |
| 24 | n := nextGen() |
| 25 | if g, ok := a.backend.(GenBackend); ok { |
| 26 | return g.StartGen(ctx, rt, n, a.gens) |
| 27 | } |
| 28 | proc, ver, err := a.backend.Start(ctx, rt) |
| 29 | // The handle goes back with an error too (P2-13): a failed start may |
| 30 | // have left its process running, and the failed-wake cleanup proves it |
| 31 | // stopped by the generation it names. |
| 32 | return proc, canonicalGen(rt, n, ver), err |
| 33 | } |
| 34 | |
| 35 | // startGenExe is startGen on the identity a wake recorded (settings spec |
| 36 | // §4.2): the same next number and the same socket choice (perGen is the |
| 37 | // switch), but the process runs id's file by path, never the live link. |
| 38 | // A failed start's handle names id too when it is a handle at all (a |
| 39 | // refusal answers none): a process it may have left running is retained by |
| 40 | // that handle (P2-13), and the generations file must keep naming the file |
| 41 | // that process runs (genIdentity lists a staged path whatever the switch). |
| 42 | func (a *Service) startGenExe(ctx context.Context, ib identityBackend, rt registry.Route, id release.StagedIdentity) (Proc, Gen, error) { |
| 43 | proc, g, err := ib.StartGenExe(ctx, rt, nextGen(), a.gens, id) |
| 44 | if err != nil && g.N != 0 { |
| 45 | g.Build, g.StagedPath, g.Digest, g.Demand = id.Version, id.StagedPath, id.Digest, id.Demand |
| 46 | } |
| 47 | return proc, g, err |
| 48 | } |
| 49 | |
| 50 | // AppSocket names host's CURRENT generation's socket, for traffic: the edge |
| 51 | // proxy, scheduled delivery, the next-due probe at a stop, and the status |
| 52 | // probe (spec §7.1). It answers for an awake or stopping entry (a stopping |
| 53 | // generation is still the one in-flight requests reach). Asleep, waking or |
| 54 | // a worker: false. |
| 55 | func (a *Service) AppSocket(host string) (string, bool) { |
| 56 | a.mu.Lock() |
| 57 | defer a.mu.Unlock() |
| 58 | e := a.entries[host] |
| 59 | if e == nil || e.state == actWaking || e.gen.Socket == "" { |
| 60 | return "", false |
| 61 | } |
| 62 | return e.gen.Socket, true |
| 63 | } |
| 64 | |
| 65 | // ServingGen is AppSocket with the generation's number and build: the edge |
| 66 | // picks a connection pool, stamps X-Carlos-Version and tags cache entries |
| 67 | // from it (spec §7.2). |
| 68 | func (a *Service) ServingGen(host string) (edge.GenInfo, bool) { |
| 69 | a.mu.Lock() |
| 70 | defer a.mu.Unlock() |
| 71 | e := a.entries[host] |
| 72 | if e == nil || e.state == actWaking || e.gen.Socket == "" { |
| 73 | return edge.GenInfo{}, false |
| 74 | } |
| 75 | return edge.GenInfo{Socket: e.gen.Socket, N: e.gen.N, Build: e.gen.Build}, true |
| 76 | } |
| 77 | |
| 78 | // CacheGen is the newest generation that has been current for host in this |
| 79 | // process. Lock-free on purpose: the edge cache calls it under its own lock, |
| 80 | // and plan 3's switch purges the cache while holding a.mu. |
| 81 | func (a *Service) CacheGen(host string) (uint64, bool) { |
| 82 | v, ok := a.cacheGens.Load(host) |
| 83 | if !ok { |
| 84 | return 0, false |
| 85 | } |
| 86 | return v.(*atomic.Uint64).Load(), true |
| 87 | } |
| 88 | |
| 89 | // noteCurrent records that generation n of host has become current. |
| 90 | func (a *Service) noteCurrent(host string, n uint64) { |
| 91 | v, _ := a.cacheGens.LoadOrStore(host, new(atomic.Uint64)) |
| 92 | v.(*atomic.Uint64).Store(n) |
| 93 | } |
| 94 | |
| 95 | // OnGenRetired registers fn for every generation that stops being current |
| 96 | // (edge.GenSource). Hooks fire only with generations on. |
| 97 | func (a *Service) OnGenRetired(fn func(host, socket string)) { |
| 98 | a.retiredMu.Lock() |
| 99 | defer a.retiredMu.Unlock() |
| 100 | a.retiredHooks = append(a.retiredHooks, fn) |
| 101 | } |
| 102 | |
| 103 | // genRetired tells every hook that g has stopped being host's current |
| 104 | // generation. The caller holds no lock. A no-op with generations off: the |
| 105 | // canonical path is shared by every generation of a host then, so closing |
| 106 | // "its" idle connections could close the next process's. |
| 107 | func (a *Service) genRetired(host string, g Gen) { |
| 108 | if !a.gens || g.Socket == "" { |
| 109 | return |
| 110 | } |
| 111 | a.retiredMu.Lock() |
| 112 | hooks := append([]func(string, string){}, a.retiredHooks...) |
| 113 | a.retiredMu.Unlock() |
| 114 | for _, fn := range hooks { |
| 115 | fn(host, g.Socket) |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | // nextDue is the sleep-time next-due probe, dialled at the instance's |
| 120 | // current generation (spec §7.1, `backend_exec.go`'s NextDue). Hibernation |
| 121 | // joins any successor first (plan 3), so at a stop the current generation |
| 122 | // is the only one. With generations off AppSocket is rt.Socket, and this is |
| 123 | // the old call. |
| 124 | func (a *Service) nextDue(ctx context.Context, e *actEntry) (int64, error) { |
| 125 | if g, ok := a.backend.(GenBackend); ok { |
| 126 | if sock, ok := a.AppSocket(e.host); ok { |
| 127 | return g.NextDueVia(ctx, e.route, sock) |
| 128 | } |
| 129 | } |
| 130 | return a.backend.NextDue(ctx, e.route) |
| 131 | } |
| 132 | |
amadan