| 1 | package rollout |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "sync" |
| 6 | "time" |
| 7 | |
| 8 | "github.com/carlosframework/platform/internal/activator" |
| 9 | "github.com/carlosframework/platform/internal/appconfig" |
| 10 | "github.com/carlosframework/platform/internal/registry" |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | ) |
| 13 | |
| 14 | // appKey names one app: one scheduler per (account, app) per box (spec §6.1). |
| 15 | type appKey struct{ acct, app string } |
| 16 | |
| 17 | // AppPolicy is the app's swap policy as this box applies it (spec §4.1). |
| 18 | type AppPolicy struct { |
| 19 | Swap activator.SwapMode |
| 20 | Concurrency int |
| 21 | } |
| 22 | |
| 23 | // DefaultPolicy is §4.1's defaults: overlap, one swap at a time. |
| 24 | func DefaultPolicy() AppPolicy { |
| 25 | return AppPolicy{Swap: activator.ModeOverlap, Concurrency: 1} |
| 26 | } |
| 27 | |
| 28 | // PolicyFromManifest reads the policy keys of an app's default bundle. A nil |
| 29 | // manifest (never configured) is the defaults. A swap value this binary does |
| 30 | // not know is hold: missing or unverifiable policy never permits an overlap |
| 31 | // (spec §4.2). |
| 32 | func PolicyFromManifest(m *appconfig.Manifest) AppPolicy { |
| 33 | p := DefaultPolicy() |
| 34 | if m == nil { |
| 35 | return p |
| 36 | } |
| 37 | switch m.Swap { |
| 38 | case "", appconfig.SwapOverlap: |
| 39 | default: |
| 40 | p.Swap = activator.ModeHold |
| 41 | } |
| 42 | if m.SwapConcurrency >= 1 { |
| 43 | p.Concurrency = m.SwapConcurrency |
| 44 | } |
| 45 | return p |
| 46 | } |
| 47 | |
| 48 | // AttemptMode is an adoption attempt's mode (spec §4.2): hold if the app |
| 49 | // policy or the build's canonical demand says hold. Any demand other than "" |
| 50 | // is read as hold, so a future demand value cannot become an overlap here. |
| 51 | func AttemptMode(p AppPolicy, demand string) activator.SwapMode { |
| 52 | if p.Swap == activator.ModeHold || demand != "" { |
| 53 | return activator.ModeHold |
| 54 | } |
| 55 | return activator.ModeOverlap |
| 56 | } |
| 57 | |
| 58 | // RecoveryMode is a recovery attempt's mode (spec §4.2): the RETAINED release |
| 59 | // demand, with the CURRENT app policy, so an app switched to hold after a |
| 60 | // build passed still gets hold on its restarts and config changes. |
| 61 | func RecoveryMode(p AppPolicy, retained string) activator.SwapMode { |
| 62 | return AttemptMode(p, retained) |
| 63 | } |
| 64 | |
| 65 | // policyEntry is one app's cached policy. fills numbers the reads begun for |
| 66 | // the app and stored is the number of the read whose result is held, so an |
| 67 | // older read that finishes after a newer one has stored leaves the newer |
| 68 | // value in place (Astra R1 Flag triage T3). failedAt is the last failed |
| 69 | // read's time, zero once a read has succeeded since. |
| 70 | type policyEntry struct { |
| 71 | p AppPolicy |
| 72 | ok bool // p was read, at at |
| 73 | at time.Time |
| 74 | failedAt time.Time |
| 75 | fills uint64 |
| 76 | stored uint64 |
| 77 | } |
| 78 | |
| 79 | // policyCache reads each app's policy at most once per ttl. A failed read |
| 80 | // keeps the last value it had; an app whose policy this box has never read |
| 81 | // gets hold, because a guess of overlap could run a migration beside a build |
| 82 | // that cannot share the database. A failed read is not tried again for |
| 83 | // policyRetryAfter, so a store that is failing costs one read per app per |
| 84 | // backoff rather than one per pass, wake and attempt (Task 11 rule 1). |
| 85 | type policyCache struct { |
| 86 | load func(ctx context.Context, acct, app string) (*appconfig.Manifest, error) |
| 87 | ttl time.Duration |
| 88 | now func() time.Time |
| 89 | mu sync.Mutex |
| 90 | m map[appKey]*policyEntry |
| 91 | } |
| 92 | |
| 93 | // policyRetryAfter is how long a failed policy read waits before the next. |
| 94 | const policyRetryAfter = 5 * time.Second |
| 95 | |
| 96 | // readBound bounds each store read the scheduler makes in line (the policy, |
| 97 | // the channel pointer, the canonical demand): a store that hangs costs a |
| 98 | // pass, a wake plan or an attempt this long, not forever (Task 11 rule 2, |
| 99 | // Astra R1 Flag triage T8 and T10). |
| 100 | const readBound = 5 * time.Second |
| 101 | |
| 102 | // get is the app's policy. A missing bundle is not an error here: |
| 103 | // appconfig.Get answers it with an empty manifest, which is the defaults. An |
| 104 | // error is never the defaults. |
| 105 | func (c *policyCache) get(ctx context.Context, acct, app string) AppPolicy { |
| 106 | k := appKey{acct, app} |
| 107 | c.mu.Lock() |
| 108 | e := c.m[k] |
| 109 | if e == nil { |
| 110 | e = &policyEntry{} |
| 111 | c.m[k] = e |
| 112 | } |
| 113 | now := c.now() |
| 114 | if e.ok && now.Sub(e.at) < c.ttl { |
| 115 | p := e.p |
| 116 | c.mu.Unlock() |
| 117 | return p |
| 118 | } |
| 119 | if !e.failedAt.IsZero() && now.Sub(e.failedAt) < policyRetryAfter { |
| 120 | p := e.lastOrHold() |
| 121 | c.mu.Unlock() |
| 122 | return p |
| 123 | } |
| 124 | e.fills++ |
| 125 | fill := e.fills |
| 126 | c.mu.Unlock() |
| 127 | lctx, cancel := context.WithTimeout(ctx, readBound) |
| 128 | m, err := c.load(lctx, acct, app) |
| 129 | cancel() |
| 130 | c.mu.Lock() |
| 131 | defer c.mu.Unlock() |
| 132 | if fill > e.stored { |
| 133 | // No newer read has stored since this one began. |
| 134 | switch { |
| 135 | case err != nil && ctx.Err() != nil: |
| 136 | // The caller's own context ended (a wake or an attempt that |
| 137 | // stopped), not the read bound: that says nothing about the |
| 138 | // store, so it holds off no other caller (Task 11 fix round 1). |
| 139 | case err != nil: |
| 140 | e.failedAt = c.now() |
| 141 | default: |
| 142 | e.p, e.ok, e.at, e.failedAt, e.stored = PolicyFromManifest(m), true, c.now(), time.Time{}, fill |
| 143 | } |
| 144 | } |
| 145 | return e.lastOrHold() |
| 146 | } |
| 147 | |
| 148 | // lastOrHold is the last policy read, or hold for an app never read. |
| 149 | func (e *policyEntry) lastOrHold() AppPolicy { |
| 150 | if e.ok { |
| 151 | return e.p |
| 152 | } |
| 153 | return AppPolicy{Swap: activator.ModeHold, Concurrency: 1} |
| 154 | } |
| 155 | |
| 156 | // Managed reports whether rt is an activator-managed exec route this |
| 157 | // scheduler moves: a hibernating exec instance or a native sidecar of a known |
| 158 | // app. Named-unit routes keep root's adoption and its privileged restart |
| 159 | // (spec §13), unit-backed routes belong to systemd, and platform routes are |
| 160 | // never adopted. |
| 161 | func Managed(rt registry.Route) bool { |
| 162 | if rt.Unit != "" || rt.Backing == registry.BackingUnit { |
| 163 | return false |
| 164 | } |
| 165 | if rt.IsSidecar() { |
| 166 | return release.SidecarAdoptable(rt) |
| 167 | } |
| 168 | return release.Adoptable(rt) && rt.Hibernate |
| 169 | } |
| 170 | |
| 171 | // Restrictable reports whether the failure book can restrict rt: Managed, |
| 172 | // with whatever status the row passes through (draining, or a move's hold: |
| 173 | // moving, landing, restoring, releasing) read as ready. A move holds the row |
| 174 | // while its probe or landing launches the instance, and a restriction is a |
| 175 | // fact about the instance and its database, not about the row's status, so |
| 176 | // it must still apply there (final review round 1, finding 2, Astra). |
| 177 | // Whether the move may launch at all is the moves package's own check. |
| 178 | func Restrictable(rt registry.Route) bool { |
| 179 | rt.Status = "ready" |
| 180 | return Managed(rt) |
| 181 | } |
| 182 | |
amadan