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 "errors"
6 "log"
7 "os"
8 "path/filepath"
9 "time"
10
11 "github.com/carlosframework/platform/internal/hostbox"
12 "github.com/carlosframework/platform/internal/hostguard"
13 "github.com/carlosframework/platform/internal/hostwrite"
14 "github.com/carlosframework/platform/internal/release"
15 "github.com/carlosframework/platform/internal/rootparticipation"
16 "github.com/carlosframework/platform/internal/rootstate"
17)
18
19type rootRestartRequests struct {
20 journal *release.RootRestarts
21 source rootRestartRequestSource
22 writes hostwrite.Config
23 sys unitSystem
24 checks rootRestartChecks
25 getenv func(string) string
26 optDir string
27}
28
29// runAdoptUnderRootRelease owns release exclusion for both call sites.
30func productionRestartRequestPass(ctx context.Context, store release.Store, writes hostwrite.Config, oldPass func() error) error {
31 return restartRequestPass(legacyRootRestart, oldPass, func() error {
32 census, err := rootparticipation.Open()
33 if err != nil {
34 return err
35 }
36 defer census.State.Close()
37 state, err := rootstate.Open("/var/lib/carlos-claims/authority/fences", false, false)
38 if err != nil {
39 return err
40 }
41 defer state.Close()
42 journal, err := rootstate.Open(release.RootRestartsDir, true, false)
43 if err != nil {
44 return err
45 }
46 defer journal.Close()
47 sys := systemdUnits{}
48 source := productionProofBindingSource(store, sys, release.OptDir())
49 source.cacheIndex = false
50 checks := productionRootRestartChecks(sys, release.OptDir())
51 checks.host = source.forHost
52 p := rootRestartRequests{journal: &release.RootRestarts{Store: journal}, source: rootRestartRequestSource{state: state, store: store, census: census.Census}, writes: writes, sys: sys, checks: checks, getenv: os.Getenv, optDir: release.OptDir()}
53 return p.run(ctx)
54 })
55}
56
57func restartRequestPass(legacy func() (bool, error), oldPass, root func() error) error {
58 old, err := legacy()
59 if err != nil {
60 return err
61 }
62 if !old {
63 return root()
64 }
65 return oldPass()
66}
67
68func (p rootRestartRequests) run(ctx context.Context) error {
69 apps, err := p.source.inventory(ctx)
70 if err != nil {
71 return err
72 }
73 retained, err := p.journal.Bindings()
74 if err != nil {
75 return err
76 }
77 census, err := p.source.census()
78 if err != nil {
79 return err
80 }
81 for _, b := range retained {
82 if b.Box != census.Box {
83 return os.ErrPermission
84 }
85 if _, ok := apps[b]; !ok {
86 apps[b] = nil
87 }
88 }
89 var failures []error
90 for b, targets := range apps {
91 body, id, readErr := p.source.store.Get(ctx, release.RestartGenKey(b.Account, b.App))
92 if errors.Is(readErr, os.ErrNotExist) {
93 body, id, readErr = nil, release.RootRestartAbsent, nil
94 }
95 scopes, _, scopeErr := release.LoadRestartScopes(ctx, p.source.store, b.Account, b.App)
96 in := release.RestartSnapshot{Complete: readErr == nil && scopeErr == nil, AppID: id, AppBody: body, Targets: targets}
97 if scopes != nil {
98 in.Scopes = *scopes
99 }
100 intent, err := p.journal.Observe(b, in)
101 if err != nil {
102 failures = append(failures, errors.Join(err, readErr, scopeErr))
103 continue
104 }
105 if intent == nil {
106 continue
107 }
108 for _, progress := range intent.Targets {
109 if progress.After != "" || progress.Failed {
110 continue
111 }
112 if err := p.consume(ctx, b, intent.ID, progress); err != nil {
113 failures = append(failures, err)
114 }
115 }
116 if err := p.journal.Finish(b, intent.ID); err != nil {
117 failures = append(failures, err)
118 }
119 }
120 return errors.Join(failures...)
121}
122
123func (p rootRestartRequests) consume(ctx context.Context, b release.RestartBinding, id string, progress release.RestartProgress) error {
124 target := progress.Target
125 in, err := p.source.installation(ctx, target.Host)
126 if err != nil {
127 return err
128 }
129 if in.Empty || in.Box != b.Box || in.Account != b.Account || in.App != b.App {
130 return os.ErrPermission
131 }
132 current, err := p.source.target(ctx, in)
133 if err != nil {
134 return err
135 }
136 if current != target {
137 return os.ErrPermission
138 }
139 // A retained census never falls through the flag-off work provider.
140 if !p.writes.Enabled {
141 return os.ErrPermission
142 }
143 work, err := p.writes.Select(ctx, target.Host, "restart")
144 if err != nil {
145 return err
146 }
147 selected, err := hostbox.Selected(work)
148 if err != nil {
149 return err
150 }
151 if selected.Box != b.Box || selected.Captured != (hostguard.Captured{Incarnation: in.Incarnation, Stamp: in.Stamp}) {
152 return os.ErrPermission
153 }
154 var before proofBinding
155 check := func() error {
156 fresh, err := p.source.target(ctx, in)
157 if err != nil {
158 return err
159 }
160 if fresh != target {
161 return os.ErrPermission
162 }
163 observed, err := p.checks.host(ctx, target.Host)
164 if err != nil {
165 return err
166 }
167 if observed.Route.Account != b.Account || observed.Route.App != b.App || observed.Route.Unit != target.Unit || observed.RecordKey+"@"+observed.RecordETag != target.Source || observed.Invocation == "" || observed.PID <= 0 || !observed.authorizedRecovery() {
168 return os.ErrPermission
169 }
170 props, err := p.sys.Show(target.Unit)
171 if err != nil {
172 return err
173 }
174 if props.ActiveState != "active" || props.MainPID != observed.PID || props.InvocationID != observed.Invocation {
175 return os.ErrPermission
176 }
177 if before.Source != "" && (observed.Source != before.Source || observed.Invocation != before.Invocation || observed.Version != before.Version || observed.Digest != before.Digest) {
178 return os.ErrPermission
179 }
180 before = observed
181 return nil
182 }
183 previous := progress.Before
184 if previous == "" {
185 if err := work.With(ctx, hostwrite.Identity{}, func() error {
186 if err := check(); err != nil {
187 return err
188 }
189 previous = before.Invocation
190 return p.journal.Begin(b, id, target, previous)
191 }); err != nil {
192 return err
193 }
194 }
195 // Re-enter admission after the durable intent. A move that starts during
196 // the write cannot turn a persisted intent into permission to restart.
197 return work.With(ctx, hostwrite.Identity{}, func() error {
198 if err := check(); err != nil {
199 return err
200 }
201 if progress.Before == "" {
202 if err := p.sys.TryRestart(target.Unit); err != nil {
203 return err
204 }
205 }
206 // An ambiguous invocation cannot be dispatched again, even after reboot.
207 opts := settleOpts{Timeout: unitSettleTimeout(p.getenv), Poll: 500 * time.Millisecond, Hold: 5 * time.Second, RequireExe: true, WantDir: filepath.Dir(release.VersionPath(p.optDir, b.Account, before.Version, b.App)), Ready: p.checks.ready(before.Route), Now: p.checks.now, Sleep: p.checks.sleep}
208 if err := settleUnit(p.sys, target.Unit, previous, opts); err != nil {
209 if progress.Before == "" && restartEffectFailed(err) {
210 props, showErr := p.sys.Show(target.Unit)
211 failed, bindingErr := p.checks.host(ctx, target.Host)
212 if showErr == nil && bindingErr == nil && props.ActiveState == "failed" && failed.Source == before.Source && failed.Version == before.Version && failed.Digest == before.Digest {
213 return errors.Join(err, p.journal.Fail(b, id, target))
214 }
215 }
216 return err
217 }
218 after, err := p.checks.host(ctx, target.Host)
219 if err != nil {
220 return err
221 }
222 fresh, err := p.source.target(ctx, in)
223 if err != nil {
224 return err
225 }
226 if fresh != target || after.Source != before.Source || after.Version != before.Version || after.Digest != before.Digest || after.Invocation == previous || after.PID <= 0 {
227 return os.ErrPermission
228 }
229 return p.journal.Complete(b, id, target, after.Invocation)
230 })
231}
232
233func logRestartRequestError(err error) {
234 if err != nil {
235 log.Printf("adopt: root restart requests: %v", err)
236 }
237}
238