amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "context"
5 "crypto/sha256"
6 "encoding/hex"
7 "encoding/json"
8 "errors"
9 "io"
10 "log"
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/hostwrite"
19 "github.com/carlosframework/platform/internal/registry"
20 "github.com/carlosframework/platform/internal/release"
21 "github.com/carlosframework/platform/internal/rootstate"
22)
23
24type rootRestartChecks struct {
25 unitShape func(context.Context, string, *registry.Route) error
26 preserveOutcome bool
27 history func() (*release.RootHistoryPass, error)
28 readLive func(string) (string, error)
29 legacy func() (bool, error)
30 legacyLog *sync.Once
31 legacyReady func(registry.Route) func() error
32 legacyStart func(string) error
33 targetDigest string
34 targetFile func(string) (*os.File, error)
35 dispatch func(context.Context, registry.Route) (bool, error)
36 resolve proofBindingResolver
37 started *sync.Map
38 attempts func() (rootstate.Store, error)
39 rollback func(context.Context, proofBinding) (proofBinding, error)
40 prepareRollback func(context.Context, proofBinding) (proofBinding, error)
41 recovered func(context.Context, proofBinding) (proofBinding, error)
42 identity hostwrite.Identity
43 effect func(func() error) error
44 verifyOnly bool
45 host func(context.Context, string) (proofBinding, error)
46 ready func(registry.Route) func() error
47 now func() time.Time
48 sleep func(time.Duration)
49}
50
51func productionRootRestartChecks(sys unitSystem, optDir string) rootRestartChecks {
52 var sourceMu sync.Mutex
53 var cached *proofBindingSource
54 var sourceErr error
55 var opened bool
56 source := func(ctx context.Context) (*proofBindingSource, error) {
57 sourceMu.Lock()
58 defer sourceMu.Unlock()
59 if opened {
60 return cached, sourceErr
61 }
62 opened = true
63 s, err := openDeployStore(ctx)
64 sourceErr = err
65 if err != nil {
66 return nil, err
67 }
68 cached = productionProofBindingSource(s, sys, optDir)
69 return cached, nil
70 }
71 return rootRestartChecks{unitShape: rootClaimRestartShape, legacy: legacyRootRestart, legacyLog: &sync.Once{}, targetFile: rootstate.OpenTrustedFile, started: &sync.Map{}, readLive: rootstate.ReadTrustedLink,
72 recovered: func(ctx context.Context, b proofBinding) (proofBinding, error) {
73 s, err := source(ctx)
74 if err != nil {
75 return proofBinding{}, err
76 }
77 return s.recovered(ctx, b)
78 },
79 history: func() (*release.RootHistoryPass, error) {
80 s, err := openDeployStore(context.Background())
81 if err != nil {
82 return nil, err
83 }
84 return openRootHistory(s, optDir)
85 },
86 dispatch: func(ctx context.Context, rt registry.Route) (bool, error) {
87 s, err := source(ctx)
88 if err != nil {
89 return false, err
90 }
91 return s.needsRestart(ctx, rt)
92 },
93 attempts: func() (rootstate.Store, error) {
94 if err := rootstate.EnsurePublicDir(release.RootHealthyDir); err != nil {
95 return nil, err
96 }
97 return rootstate.Open(rootRestartState, true, false)
98 },
99 prepareRollback: func(ctx context.Context, b proofBinding) (proofBinding, error) {
100 s, err := source(ctx)
101 if err != nil {
102 return proofBinding{}, err
103 }
104 return s.prepareRollback(ctx, b)
105 },
106 rollback: func(ctx context.Context, b proofBinding) (proofBinding, error) {
107 s, err := source(ctx)
108 if err != nil {
109 return proofBinding{}, err
110 }
111 return s.rollback(ctx, b)
112 },
113 resolve: func(ctx context.Context, rt registry.Route) (proofBinding, error) {
114 s, err := source(ctx)
115 if err != nil {
116 return proofBinding{}, err
117 }
118 return s.resolve(ctx, rt)
119 },
120 host: func(ctx context.Context, host string) (proofBinding, error) {
121 s, err := source(ctx)
122 if err != nil {
123 return proofBinding{}, err
124 }
125 return s.forHost(ctx, host)
126 },
127 ready: func(rt registry.Route) func() error { return proofRouteReady(rt, 2*time.Second) },
128 now: time.Now, sleep: time.Sleep,
129 }
130}
131
132func (c rootRestartChecks) runningVersion(ctx context.Context, rt registry.Route) (string, error) {
133 if rt.Unit == "" || rt.Unit == activator.UnitName(rt.Host) {
134 return "", nil
135 }
136 // Merge ruling 6: without a census the binding never chooses an
137 // effect, and the rollback target is one. Main's cur stands.
138 legacy, err := c.allowLegacy()
139 if err != nil {
140 return "", err
141 }
142 if legacy {
143 return "", nil
144 }
145 b, err := c.resolve(ctx, rt)
146 if err != nil {
147 return "", err
148 }
149 if b.PID > 0 {
150 return b.Version, nil
151 }
152 return "", nil
153}
154
155func checkedUnitRestart(sys unitSystem, optDir string, getenv func(string) string, settled func(registry.Route) error, checks ...rootRestartChecks) func(registry.Route) error {
156 c := productionRootRestartChecks(sys, optDir)
157 if len(checks) == 1 {
158 c = checks[0]
159 }
160 return func(candidate registry.Route) error {
161 c := c
162 legacy, err := c.allowLegacy()
163 if err != nil {
164 return err
165 }
166 if legacy {
167 c.noteLegacy()
168 return legacyCheckedUnitRestart(sys, optDir, getenv, candidate, settled, c)
169 }
170 if err := c.checkUnitShape(context.Background(), candidate.Host, &candidate); err != nil {
171 return err
172 }
173 if c.attempts == nil {
174 return os.ErrPermission
175 }
176 s, err := c.attempts()
177 if err != nil {
178 return err
179 }
180 defer s.Close()
181 unlock, err := s.Lock(context.Background(), rollbackHostKey(candidate.Host)+".lock")
182 if err != nil {
183 return err
184 }
185 defer unlock()
186 binding, err := c.resolve(context.Background(), candidate)
187 if err != nil {
188 return err
189 }
190 before, err := sys.Show(binding.Route.Unit)
191 if err != nil {
192 return err
193 }
194 if before.ActiveState == "inactive" {
195 return errUnitStopped
196 }
197 if !pointerBoundUnit(before) && (before.ActiveState != "active" || before.MainPID <= 0 || before.MainPID != binding.PID || before.InvocationID != binding.Invocation) {
198 return os.ErrPermission
199 }
200 target, err := c.live(optDir, candidate.Host)
201 if err != nil {
202 return err
203 }
204 rel, err := filepath.Rel(filepath.Join(optDir, "versions"), target)
205 if err != nil {
206 return err
207 }
208 version := strings.Split(rel, string(filepath.Separator))[0]
209 if !release.ValidVersion(version) || target != release.VersionPath(optDir, binding.Route.Account, version, binding.Route.App) || c.targetFile == nil {
210 return os.ErrPermission
211 }
212 f, err := c.targetFile(target)
213 if err != nil {
214 return err
215 }
216 hash := sha256.New()
217 _, err = io.Copy(hash, f)
218 f.Close()
219 if err != nil {
220 return err
221 }
222 targetDigest := hex.EncodeToString(hash.Sum(nil))
223 c.targetDigest = targetDigest
224 captured := binding
225 if pointerBoundUnit(before) {
226 previous, e := filepath.EvalSymlinks(filepath.Join(optDir, "previous", candidate.Host))
227 if e == nil && previous != target {
228 rel, e := filepath.Rel(filepath.Join(optDir, "versions"), previous)
229 if e != nil {
230 return e
231 }
232 version := strings.Split(rel, string(filepath.Separator))[0]
233 if !release.ValidVersion(version) || previous != release.VersionPath(optDir, binding.Route.Account, version, binding.Route.App) {
234 return os.ErrPermission
235 }
236 f, e := c.targetFile(previous)
237 if e != nil {
238 return e
239 }
240 h := sha256.New()
241 _, e = io.Copy(h, f)
242 f.Close()
243 if e != nil {
244 return e
245 }
246 captured.Version, captured.Digest = version, hex.EncodeToString(h.Sum(nil))
247 captured, e = c.prepareRollback(context.Background(), captured)
248 if e != nil {
249 return e
250 }
251 } else if e != nil && !errors.Is(e, os.ErrNotExist) {
252 return e
253 }
254 }
255 if err := writeRollbackState(s, rollbackHostKey(candidate.Host)+".previous", captured); err != nil {
256 return err
257 }
258 attempt, err := historyRestartAttempt(s, captured, c, target, getenv)
259 if err != nil {
260 return err
261 }
262 if c.started != nil {
263 c.started.Store(candidate.Host, attempt.ID)
264 }
265 err = restartBoundUnit(sys, optDir, getenv, binding, false, settled, c)
266 if err == nil {
267 return settleRollbackAttempt(s, attempt)
268 }
269 if errors.Is(err, errUnitStopped) {
270 return errors.Join(err, settleRollbackAttempt(s, attempt))
271 }
272 if errors.Is(err, errRestartBinding) && !restartEffectFailed(err) {
273 return errors.Join(err, settleRollbackAttempt(s, attempt))
274 }
275 attempt.Failed = true
276 var notUp *release.UnitNotUpError
277 if !errors.As(err, &notUp) {
278 err = errors.Join(&release.UnitNotUpError{Unit: binding.Route.Unit, Reason: err.Error()}, err)
279 }
280 return errors.Join(err, writeRollbackState(s, rollbackJournalKey(candidate.Host), attempt))
281 }
282}
283
284func restartBoundUnit(sys unitSystem, optDir string, getenv func(string) string, binding proofBinding, rollback bool, settled func(registry.Route) error, c rootRestartChecks) error {
285 rt := binding.Route
286 if binding.Source == "" || !binding.Network || c.ready == nil {
287 return errRestartBinding
288 }
289 ready := c.ready(rt)
290 if ready == nil {
291 return errRestartBinding
292 }
293 before, err := sys.Show(rt.Unit)
294 if err != nil {
295 return errors.Join(errRestartBinding, err)
296 }
297 if before.ExecPath != proofExecPath(binding, optDir) {
298 return errRestartBinding
299 }
300 if before.ActiveState == "inactive" {
301 return errUnitStopped
302 }
303 want, err := c.live(optDir, rt.Host)
304 if err != nil {
305 return err
306 }
307 effect := c.effect
308 if effect == nil {
309 effect = func(fn func() error) error { return fn() }
310 }
311 if c.verifyOnly {
312 before.InvocationID = ""
313 } else if err := effect(func() error {
314 if rollback {
315 current, err := c.rollback(context.Background(), binding)
316 if err != nil || current.Source != binding.Source {
317 return errRestartBinding
318 }
319 if err := sys.ResetFailed(rt.Unit); err != nil {
320 return err
321 }
322 if err := sys.Restart(rt.Unit); err != nil {
323 return err
324 }
325 } else {
326 if !pointerBoundUnit(before) && (before.ActiveState != "active" || before.MainPID != binding.PID || before.InvocationID != binding.Invocation) {
327 return errRestartBinding
328 }
329 if before.MainPID > 0 {
330 start, err := sys.StartTime(before.MainPID)
331 if err != nil || start != binding.Start {
332 return errRestartBinding
333 }
334 }
335 current, err := c.host(context.Background(), rt.Host)
336 if err != nil || current.Source != binding.Source || current.Version != binding.Version || current.Digest != binding.Digest || current.PID != binding.PID || current.Invocation != binding.Invocation {
337 return errRestartBinding
338 }
339 if pointerBoundUnit(before) {
340 if err := sys.ResetFailed(rt.Unit); err != nil {
341 return err
342 }
343 return sys.Restart(rt.Unit)
344 }
345 if err := sys.TryRestart(rt.Unit); err != nil {
346 return err
347 }
348 }
349 return nil
350 }); err != nil {
351 return err
352 }
353 var observed proofBinding
354 observe := func() (proofBinding, error) {
355 if rollback && c.recovered != nil {
356 return c.recovered(context.Background(), binding)
357 }
358 return c.host(context.Background(), rt.Host)
359 }
360 var bindingErr error
361 checkedReady := func() error {
362 first, err := observe()
363 if err != nil || first.Source != binding.Source || want != release.VersionPath(optDir, first.Route.Account, first.Version, first.Route.App) {
364 bindingErr = errors.Join(errRestartBinding, err)
365 return bindingErr
366 }
367 if c.targetDigest != "" && first.Digest != c.targetDigest {
368 bindingErr = errRestartBinding
369 return bindingErr
370 }
371 if rollback && (first.Version != binding.Version || first.Digest != binding.Digest) {
372 bindingErr = errRestartBinding
373 return bindingErr
374 }
375 if err := ready(); err != nil {
376 return err
377 }
378 last, err := observe()
379 if err != nil || last.Source != first.Source || last.Version != first.Version || last.Digest != first.Digest || last.Invocation != first.Invocation || last.PID != first.PID || last.Start != first.Start {
380 bindingErr = errors.Join(errRestartBinding, err)
381 return bindingErr
382 }
383 observed = last
384 return nil
385 }
386 err = settleUnit(sys, rt.Unit, before.InvocationID, settleOpts{
387 Timeout: unitSettleTimeout(getenv), Poll: 500 * time.Millisecond, Hold: 5 * time.Second,
388 WantDir: filepath.Dir(want), Ready: checkedReady, Now: c.now, Sleep: c.sleep, RequireExe: true,
389 })
390 if err != nil {
391 if bindingErr != nil && !restartEffectFailed(err) {
392 if c.preserveOutcome {
393 return errors.Join(err, bindingErr)
394 }
395 return bindingErr
396 }
397 return err
398 }
399 current, err := observe()
400 if err != nil {
401 return errors.Join(errRestartBinding, err)
402 }
403 if current.Invocation != observed.Invocation || current.PID != observed.PID || current.Start != observed.Start || current.Source != binding.Source || want != release.VersionPath(optDir, current.Route.Account, current.Version, current.Route.App) || (rollback && (current.Version != binding.Version || current.Digest != binding.Digest)) {
404 return errRestartBinding
405 }
406 if settled != nil {
407 if err := effect(func() error { return settled(binding.Route) }); err != nil {
408 log.Print(err)
409 }
410 }
411 return nil
412}
413
414func (c rootRestartChecks) live(optDir, host string) (string, error) {
415 if c.readLive != nil {
416 return c.readLive(release.LiveLink(optDir, host))
417 }
418 return filepath.EvalSymlinks(release.LiveLink(optDir, host))
419}
420
421func rollbackUnitRestart(reg *registry.Registry, sys unitSystem, optDir string, getenv func(string) string, checks ...rootRestartChecks) func(string) error {
422 c := productionRootRestartChecks(sys, optDir)
423 if len(checks) == 1 {
424 c = checks[0]
425 }
426
427 return func(host string) (result error) {
428 c := c
429 rt, routeErr := reg.Get(host)
430 if err := c.checkUnitShape(context.Background(), host, rt); err != nil {
431 return err
432 }
433 if rt != nil && (rt.Unit == "" || (c.legacyStart != nil && rt.Hibernate)) {
434 return nil
435 }
436 if rt != nil && rt.Unit == activator.UnitName(host) {
437 if c.legacyStart != nil {
438 return c.legacyStart(rt.Unit)
439 }
440 return legacyRollbackRestart(sys, optDir, getenv, *rt, c)
441 }
442 legacy, err := c.allowLegacy()
443 if err != nil {
444 return err
445 }
446 if legacy {
447 c.noteLegacy()
448 if routeErr != nil || rt == nil {
449 return nil
450 }
451 if c.legacyStart != nil {
452 return c.legacyStart(rt.Unit)
453 }
454 return legacyRollbackRestart(sys, optDir, getenv, *rt, c)
455 }
456 if c.attempts == nil || c.rollback == nil {
457 return os.ErrPermission
458 }
459 s, err := c.attempts()
460 if err != nil {
461 return err
462 }
463 defer s.Close()
464 unlock, err := s.Lock(context.Background(), rollbackHostKey(host)+".lock")
465 if err != nil {
466 return err
467 }
468 defer unlock()
469 attempt, binding, err := loadRollback(s, host, c.now())
470 if attempt.ID != "" {
471 defer func() {
472 result = errors.Join(result, settleRollbackAttempt(s, attempt))
473 }()
474 }
475 if err != nil {
476 if !errors.Is(err, os.ErrNotExist) && !attempt.Settled && (attempt.ID == "" || !c.now().After(attempt.Deadline.Add(rollbackMargin))) {
477 return err
478 }
479 raw, readErr := s.Read(rollbackHostKey(host) + ".previous")
480 if errors.Is(readErr, os.ErrNotExist) && c.host != nil {
481 binding, readErr = c.host(context.Background(), host)
482 } else if readErr == nil && json.Unmarshal(raw, &binding) != nil {
483 readErr = os.ErrPermission
484 }
485 if readErr != nil || binding.Route.Host != host {
486 return errors.Join(err, readErr)
487 }
488 before, showErr := sys.Show(binding.Route.Unit)
489 if showErr != nil || before.ActiveState != "failed" || before.MainPID != 0 {
490 return errors.Join(err, showErr)
491 }
492 }
493 c.targetDigest = binding.Digest
494 binding, err = c.rollback(context.Background(), binding)
495 if err != nil {
496 return err
497 }
498 before, err := sys.Show(binding.Route.Unit)
499 if err != nil {
500 return err
501 }
502 if before.ActiveState == "inactive" {
503 return errUnitStopped
504 }
505 if before.ActiveState != "failed" && before.MainPID != 0 && !attempt.Failed {
506 return os.ErrPermission
507 }
508 if attempt.Recovering && before.ActiveState == "active" && before.MainPID > 0 {
509 c.verifyOnly = true
510 }
511 attempt.Recovering = true
512 if err := writeRollbackState(s, rollbackJournalKey(host), attempt); err != nil {
513 return err
514 }
515 if err := restartBoundUnit(sys, optDir, getenv, binding, true, nil, c); err != nil {
516 return err
517 }
518 attempt.Verified = true
519 return c.settleHistory(context.Background(), attempt, true)
520 }
521
522}
523
524func rootStartRoute(reg *registry.Registry, optDir string, checks rootRestartChecks) func(string) error {
525 checks.legacyStart = systemctlStartUnit
526 return rollbackUnitRestart(reg, systemdUnits{}, optDir, os.Getenv, checks)
527}
528
529func rootSelectedRestart(reg *registry.Registry, sys unitSystem, optDir string, getenv func(string) string, settled func(registry.Route) error, checks rootRestartChecks, rollback bool) hostwrite.Callback {
530 return func(ctx context.Context, selected hostwrite.Selection) error {
531 if selected.Work == nil {
532 return hostwrite.ErrUnavailable
533 }
534 c := checks
535 c.identity = selected.Identity
536 c.effect = func(fn func() error) error { return selected.Work.With(ctx, selected.Identity, fn) }
537 if rollback {
538 return rollbackUnitRestart(reg, sys, optDir, getenv, c)(selected.Host)
539 }
540 return rootRestartRouteWith(reg, sys, optDir, getenv, settled, c)(selected.Host)
541 }
542}
543
544func rootRestartRouteWith(reg *registry.Registry, sys unitSystem, optDir string, getenv func(string) string, settled func(registry.Route) error, checks rootRestartChecks) func(string) error {
545 restart := checkedUnitRestart(sys, optDir, getenv, settled, checks)
546 return func(host string) error {
547 rt, err := reg.Get(host)
548 if shapeErr := checks.checkUnitShape(context.Background(), host, rt); shapeErr != nil {
549 return shapeErr
550 }
551 if err != nil || rt == nil || rt.Unit == "" {
552 return restartOrdinaryUnit(sys, host)
553 }
554 if rt.Unit == activator.UnitName(host) {
555 return legacyCheckedUnitRestart(sys, optDir, getenv, *rt, settled, checks)
556 }
557 legacy, err := checks.allowLegacy()
558 if err != nil {
559 return err
560 }
561 if legacy {
562 checks.noteLegacy()
563 return legacyCheckedUnitRestart(sys, optDir, getenv, *rt, settled, checks)
564 }
565 if checks.dispatch != nil {
566 need, err := checks.dispatch(context.Background(), *rt)
567 if err != nil {
568 return err
569 }
570 if !need {
571 return restartVerifiedOrdinaryUnit(sys, optDir, host, checks.targetFile)
572 }
573 }
574 err = restart(*rt)
575 if errors.Is(err, errUnitStopped) {
576 return nil
577 }
578 return err
579 }
580}
581
582func restartOrdinaryUnit(sys unitSystem, host string) error {
583 if err := sys.TryRestart(activator.UnitName(host)); err != nil {
584 log.Print(err)
585 }
586 return nil
587}
588
589func restartVerifiedOrdinaryUnit(sys unitSystem, optDir, host string, open func(string) (*os.File, error)) error {
590 unit := activator.UnitName(host)
591 p, err := sys.Show(unit)
592 if err != nil || p.Name != unit || p.ExecPath != release.LiveLink(optDir, host) || p.ActiveState != "active" || open == nil {
593 return nil
594 }
595 file, err := open(p.FragmentPath)
596 if err != nil {
597 return err
598 }
599 file.Close()
600 if err := sys.TryRestart(unit); err != nil {
601 log.Print(err)
602 }
603 return nil
604}
605