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 "encoding/json"
6 "errors"
7 "fmt"
8 "io"
9 "os"
10 "os/exec"
11 "path/filepath"
12 "strconv"
13 "sync"
14 "time"
15
16 "github.com/carlosframework/platform/internal/hostbox"
17 "github.com/carlosframework/platform/internal/hostclaim"
18 "github.com/carlosframework/platform/internal/hostrelocate"
19 "github.com/carlosframework/platform/internal/hostwrite"
20 "github.com/carlosframework/platform/internal/instancefs"
21 "github.com/carlosframework/platform/internal/moves"
22 "github.com/carlosframework/platform/internal/release"
23)
24
25// relocationAttemptJournal is one operation's probe journal, at
26// <DataDir>/relocate/ops/<op>/attempts.json (ruling 41, keyed by op per
27// ruling 44), written whole and renamed. The claim's probe_attempts.<box>
28// is the record attempt k; the journal binds each admitted session to one
29// k, keeps each k's first app start, and each k's current start. Every
30// field has its own JSON name (finding 14).
31type relocationAttemptJournal struct {
32 V int `json:"v"`
33 Host string `json:"host"`
34 Op string `json:"op"`
35 Opened map[string]time.Time `json:"opened"` // "k" -> the attempt's first app start
36 Sessions map[string]int `json:"sessions"` // "session" -> k
37 Starts map[string]relocationStart `json:"starts"` // "k" -> the attempt's current start
38}
39
40// relocationStart is one start's reservation, bound to the admitted
41// session and generation root's reader validates (Plan 11: the exact
42// operation, phase, generation, stamp and session).
43type relocationStart struct {
44 Session uint64 `json:"session"`
45 Generation string `json:"generation"`
46 Stamp uint64 `json:"stamp"`
47 N int `json:"n"`
48 Attempt hostclaim.WakeRetryAttempt `json:"attempt"`
49 State string `json:"state"` // reserved, failed, healthy
50}
51
52func relocationAttemptsPath(dataDir, op string) (string, error) {
53 dir, err := relocationOpDir(dataDir, op)
54 if err != nil {
55 return "", err
56 }
57 return filepath.Join(dir, "attempts.json"), nil
58}
59
60// readRelocationAttempts is the agent's read of its own journal: a
61// missing file is the journal's first state (OpenProbe creates it).
62func readRelocationAttempts(dataDir, host, op string) (relocationAttemptJournal, error) {
63 path, err := relocationAttemptsPath(dataDir, op)
64 if err != nil {
65 return relocationAttemptJournal{}, err
66 }
67 raw, err := os.ReadFile(path)
68 if errors.Is(err, os.ErrNotExist) {
69 return relocationAttemptJournal{V: 1, Host: host, Op: op, Opened: map[string]time.Time{}, Sessions: map[string]int{}, Starts: map[string]relocationStart{}}, nil
70 }
71 if err != nil {
72 return relocationAttemptJournal{}, err
73 }
74 return decodeRelocationAttempts(raw, host, op)
75}
76
77func decodeRelocationAttempts(raw []byte, host, op string) (relocationAttemptJournal, error) {
78 var j relocationAttemptJournal
79 if err := json.Unmarshal(raw, &j); err != nil || j.V != 1 || j.Host != host || j.Op != op || j.Opened == nil || j.Sessions == nil || j.Starts == nil {
80 return relocationAttemptJournal{}, fmt.Errorf("%w: relocation attempt journal %s", hostbox.ErrChanged, op)
81 }
82 return j, nil
83}
84
85// relocationAttemptsLimit bounds root's read of the journal: three record
86// attempts, each with one start and one session binding, is well under a
87// kibibyte.
88const relocationAttemptsLimit = 64 << 10
89
90// readRelocationAttemptsBeneath is root's read of the agent-written
91// journal (fix round 1, I1). The file and its directories belong to the
92// agent's uid, which may replace them at will, so root opens it the way
93// it opens every agent-writable path on main: beneath the data root with
94// openat2 (RESOLVE_BENEATH | RESOLVE_NO_SYMLINKS, instancefs), O_NONBLOCK
95// so a FIFO cannot hold root under the host guard, a regular file only,
96// the size capped before and during the read, and the context honoured.
97// A missing journal is a refusal, never a fresh one: root takes nothing
98// the journal does not hold.
99func readRelocationAttemptsBeneath(ctx context.Context, dataDir, host, op string) (relocationAttemptJournal, error) {
100 if err := ctx.Err(); err != nil {
101 return relocationAttemptJournal{}, err
102 }
103 if dataDir == "" || !validOpID(op) {
104 return relocationAttemptJournal{}, ErrClaimStopUnavailable
105 }
106 f, err := instancefs.OpenBeneath(dataDir, filepath.Join("relocate", "ops", op, "attempts.json"), os.O_RDONLY, 0)
107 if err != nil {
108 return relocationAttemptJournal{}, fmt.Errorf("%w: relocation attempt journal %s: %w", hostbox.ErrChanged, op, err)
109 }
110 defer f.Close()
111 st, err := f.Stat()
112 if err != nil {
113 return relocationAttemptJournal{}, err
114 }
115 if !st.Mode().IsRegular() || st.Size() > relocationAttemptsLimit {
116 return relocationAttemptJournal{}, fmt.Errorf("%w: relocation attempt journal %s is not a regular file within %d bytes", hostbox.ErrChanged, op, relocationAttemptsLimit)
117 }
118 raw, err := io.ReadAll(io.LimitReader(f, relocationAttemptsLimit+1))
119 if err != nil {
120 return relocationAttemptJournal{}, err
121 }
122 if len(raw) > relocationAttemptsLimit {
123 return relocationAttemptJournal{}, fmt.Errorf("%w: relocation attempt journal %s grew past %d bytes", hostbox.ErrChanged, op, relocationAttemptsLimit)
124 }
125 if err := ctx.Err(); err != nil {
126 return relocationAttemptJournal{}, err
127 }
128 return decodeRelocationAttempts(raw, host, op)
129}
130
131func (j relocationAttemptJournal) write(dataDir string) error {
132 path, err := relocationAttemptsPath(dataDir, j.Op)
133 if err != nil {
134 return err
135 }
136 raw, err := json.Marshal(j)
137 if err != nil {
138 return err
139 }
140 return writeRelocationJournal(path, raw)
141}
142
143// relocationProbeAttempts is region HA's OperationAttempts for a probe
144// admitted under relocating/probing (claim Plan 11 R1, D2), modelled on
145// movesProbeAttempts without the record. Every method refuses
146// (hostbox.ErrChanged) once the claim no longer names the admitted writer
147// at probing, unstuck, with k = probe_attempts.<box> in 1..MaxProbeAttempts.
148type relocationProbeAttempts struct {
149 Claims release.Store
150 DataDir string
151 Box string
152 NewID func() string
153 Now func() time.Time
154 Settle, Ready time.Duration
155 mu sync.Mutex
156 current map[string]boundAttempt // host\x00session
157}
158
159var _ OperationAttempts = (*relocationProbeAttempts)(nil)
160
161func (p *relocationProbeAttempts) bound(ctx context.Context, s OperationStart) (int, error) {
162 if p == nil || p.Claims == nil || p.DataDir == "" || p.Box == "" || p.NewID == nil || p.Now == nil {
163 return 0, hostwrite.ErrUnavailable
164 }
165 a := s.Admitted
166 if a.Claim == nil || a.Claim.Holder == nil || a.Claim.Op.Kind != hostclaim.KindRelocating || !s.Policy.Probe {
167 return 0, hostbox.ErrUnsupported
168 }
169 h := a.Claim.Holder
170 if s.Owner.Host != a.Host || s.Owner.Session != h.Session || s.Owner.Stamp != h.Stamp || s.Owner.Incarnation != h.Incarnation || s.Owner.Box != p.Box {
171 return 0, hostbox.ErrChanged
172 }
173 fresh, err := hostclaim.Read(ctx, p.Claims, a.Host)
174 if err != nil {
175 return 0, err
176 }
177 if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseRelocatingProbing || fresh.Claim.Op.Journal["stuck"] != "" || fresh.Claim.Op.Exec != p.Box {
178 return 0, hostbox.ErrChanged
179 }
180 k, err := strconv.Atoi(fresh.Claim.Op.Journal["probe_attempts."+p.Box])
181 if err != nil || k < 1 || k > hostclaim.MaxProbeAttempts {
182 return 0, hostbox.ErrChanged
183 }
184 return k, nil
185}
186
187func relocationSessionKey(s OperationStart) string {
188 return s.Admitted.Host + "\x00" + strconv.FormatUint(s.Owner.Session, 10)
189}
190
191// bind checks, and with write records, the session's record attempt.
192func (j relocationAttemptJournal) bind(session uint64, k int, write bool) error {
193 key := strconv.FormatUint(session, 10)
194 got, ok := j.Sessions[key]
195 switch {
196 case ok && got == k:
197 return nil
198 case ok:
199 return fmt.Errorf("%w: session %d is bound to attempt %d, not %d", hostbox.ErrChanged, session, got, k)
200 case !write:
201 return fmt.Errorf("%w: session %d is bound to no attempt", hostbox.ErrChanged, session)
202 }
203 j.Sessions[key] = k
204 return nil
205}
206
207func (p *relocationProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, proposed time.Time) (time.Time, error) {
208 k, err := p.bound(ctx, s)
209 if err != nil {
210 return time.Time{}, err
211 }
212 if proposed.IsZero() {
213 return time.Time{}, hostwrite.ErrUnavailable
214 }
215 p.mu.Lock()
216 defer p.mu.Unlock()
217 j, err := readRelocationAttempts(p.DataDir, s.Admitted.Host, s.Admitted.Claim.Op.ID)
218 if err != nil {
219 return time.Time{}, err
220 }
221 if err := j.bind(s.Owner.Session, k, true); err != nil {
222 return time.Time{}, err
223 }
224 key := strconv.Itoa(k)
225 opened, ok := j.Opened[key]
226 if !ok {
227 opened = proposed.UTC()
228 j.Opened[key] = opened
229 }
230 return opened, j.write(p.DataDir)
231}
232
233func (p *relocationProbeAttempts) Reserve(ctx context.Context, s OperationStart, n int) error {
234 k, err := p.bound(ctx, s)
235 if err != nil {
236 return err
237 }
238 digest, err := ClaimLaunchDigest(s.Inputs)
239 if err != nil {
240 return err
241 }
242 p.mu.Lock()
243 defer p.mu.Unlock()
244 j, err := readRelocationAttempts(p.DataDir, s.Admitted.Host, s.Admitted.Claim.Op.ID)
245 if err != nil {
246 return err
247 }
248 if err := j.bind(s.Owner.Session, k, n == 0); err != nil {
249 return err
250 }
251 key := strconv.Itoa(k)
252 st, ok := j.Starts[key]
253 same := ok && st.Session == s.Owner.Session
254 h := s.Admitted.Claim.Holder
255 switch {
256 case n < 0:
257 return hostbox.ErrChanged
258 case n == 0 && same:
259 if st.N != 0 || st.State != probeReserved || !sameStartOwner(st.Attempt, s, digest) || st.Generation != h.Generation || st.Stamp != h.Stamp {
260 return hostbox.ErrChanged
261 }
262 case n == 0:
263 st = relocationStart{Session: s.Owner.Session, Generation: h.Generation, Stamp: h.Stamp, State: probeReserved}
264 default:
265 if !same || st.N != n-1 || st.State != probeFailed || st.Attempt.LaunchInputs != digest {
266 return hostbox.ErrChanged
267 }
268 st = relocationStart{Session: s.Owner.Session, Generation: h.Generation, Stamp: h.Stamp, N: n, State: probeReserved}
269 }
270 if st.Attempt.ID == "" {
271 o := s.Owner
272 st.Attempt = hostclaim.WakeRetryAttempt{ID: p.NewID(), Box: o.Box, Enrollment: o.Enrollment, BootID: o.BootID, ControllerID: o.ControllerID, Session: o.Session, InvocationID: p.NewID(), LaunchInputs: digest}
273 if st.Attempt.ID == "" || st.Attempt.InvocationID == "" || st.Attempt.ID == st.Attempt.InvocationID {
274 return hostwrite.ErrUnavailable
275 }
276 j.Starts[key] = st
277 }
278 if err := j.write(p.DataDir); err != nil {
279 return err
280 }
281 if p.current == nil {
282 p.current = map[string]boundAttempt{}
283 }
284 p.current[relocationSessionKey(s)] = boundAttempt{k: k, attempt: st.Attempt}
285 return nil
286}
287
288func (p *relocationProbeAttempts) CurrentAttempt(s OperationStart) hostclaim.WakeRetryAttempt {
289 if p == nil {
290 return hostclaim.WakeRetryAttempt{}
291 }
292 p.mu.Lock()
293 cur, ok := p.current[relocationSessionKey(s)]
294 p.mu.Unlock()
295 if !ok {
296 return hostclaim.WakeRetryAttempt{}
297 }
298 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
299 defer cancel()
300 if k, err := p.bound(ctx, s); err != nil || k != cur.k {
301 return hostclaim.WakeRetryAttempt{}
302 }
303 return cur.attempt
304}
305
306func (p *relocationProbeAttempts) Failed(ctx context.Context, s OperationStart, n int) error {
307 recovery, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
308 defer cancel()
309 return p.settle(recovery, s, n, probeFailed)
310}
311
312func (p *relocationProbeAttempts) settle(ctx context.Context, s OperationStart, n int, state string) error {
313 k, err := p.bound(ctx, s)
314 if err != nil {
315 return err
316 }
317 p.mu.Lock()
318 defer p.mu.Unlock()
319 j, err := readRelocationAttempts(p.DataDir, s.Admitted.Host, s.Admitted.Claim.Op.ID)
320 if err != nil {
321 return err
322 }
323 if err := j.bind(s.Owner.Session, k, false); err != nil {
324 return err
325 }
326 key := strconv.Itoa(k)
327 st, ok := j.Starts[key]
328 cur := p.current[relocationSessionKey(s)]
329 if !ok || st.Session != s.Owner.Session || st.N != n || st.State != probeReserved || cur.k != k || st.Attempt != cur.attempt {
330 return hostbox.ErrChanged
331 }
332 st.State = state
333 j.Starts[key] = st
334 return j.write(p.DataDir)
335}
336
337// Healthy is movesProbeAttempts.Healthy's settle and second request
338// (moves §7.7), which the relocation probe shares (spec §5.8).
339func (p *relocationProbeAttempts) Healthy(ctx context.Context, s OperationStart, n int, cmd *exec.Cmd, again func(context.Context) error) error {
340 if _, err := p.bound(ctx, s); err != nil {
341 return err
342 }
343 if again == nil || !probeAlive(cmd) {
344 return hostbox.ErrChanged
345 }
346 settle, ready := p.Settle, p.Ready
347 if settle <= 0 {
348 settle = time.Duration(moves.SettleSeconds * float64(time.Second))
349 }
350 if ready <= 0 {
351 ready = time.Duration(moves.ProbeSeconds * float64(time.Second))
352 }
353 timer := time.NewTimer(settle)
354 select {
355 case <-ctx.Done():
356 timer.Stop()
357 return ctx.Err()
358 case <-timer.C:
359 }
360 if !probeAlive(cmd) {
361 return fmt.Errorf("%w: the app exited during the settle", hostbox.ErrChanged)
362 }
363 asked, cancel := context.WithTimeout(ctx, ready)
364 err := again(asked)
365 cancel()
366 if err != nil {
367 return err
368 }
369 if !probeAlive(cmd) {
370 return fmt.Errorf("%w: the app exited after the second request", hostbox.ErrChanged)
371 }
372 return p.settle(ctx, s, n, probeHealthy)
373}
374
375// relocationPassedProbe is hostrelocate.Effects.PassedProbe (final review
376// A2): the attempt journal's healthy start for the claim's holder session,
377// the state Healthy settles once the probe passed, read as the agent
378// reads its own journal. The start must be the executor's record attempt
379// k, bound to the holder's session, generation and stamp, so a start of
380// another attempt or session is never taken for this one. After P's
381// opening the journal exists (OpenProbe wrote it before the app first
382// started), so a missing journal is a refusal like an unreadable one,
383// never "no probe passed": the worker's deadlines decide such an
384// operation, not an absence.
385func relocationPassedProbe(dataDir string) func(context.Context, hostclaim.Observed) (uint64, bool, error) {
386 return func(_ context.Context, o hostclaim.Observed) (uint64, bool, error) {
387 c := o.Claim
388 if dataDir == "" || c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRelocating || c.Op.Phase != hostclaim.PhaseRelocatingProbing || !hostclaim.PreparedOpened(c) {
389 return 0, false, ErrClaimStopUnavailable
390 }
391 h := c.Holder
392 if h.WriterBox != c.Op.Exec || h.Session == 0 {
393 return 0, false, hostbox.ErrChanged
394 }
395 k, err := strconv.Atoi(c.Op.Journal["probe_attempts."+c.Op.Exec])
396 if err != nil || k < 1 || k > hostclaim.MaxProbeAttempts {
397 return 0, false, hostbox.ErrChanged
398 }
399 path, err := relocationAttemptsPath(dataDir, c.Op.ID)
400 if err != nil {
401 return 0, false, err
402 }
403 raw, err := os.ReadFile(path)
404 if err != nil {
405 return 0, false, fmt.Errorf("%w: the attempt journal of %s after P's opening: %w", hostrelocate.ErrRefused, c.Op.ID, err)
406 }
407 j, err := decodeRelocationAttempts(raw, o.Host, c.Op.ID)
408 if err != nil {
409 return 0, false, err
410 }
411 st, ok := j.Starts[strconv.Itoa(k)]
412 if !ok || j.Sessions[strconv.FormatUint(h.Session, 10)] != k || st.Session != h.Session || st.Generation != h.Generation || st.Stamp != h.Stamp || st.Attempt.Box != c.Op.Exec || st.State != probeHealthy {
413 return 0, false, nil
414 }
415 return h.Session, true, nil
416 }
417}
418
419// relocationAttemptReader is root's authoritative attempt reader for kind
420// relocating (ruling 48; claim Plan 11 "install and test root-owned
421// authoritative attempt readers"): it reads the operation's durable
422// journal and returns the reserved start only for exactly the observed
423// op at probing, its executor's attempt k, the holder's session,
424// generation and stamp (the journaled install stamp). Root's authorize
425// then binds the reservation, the launch-input digest and the peer.
426//
427// The journal is the agent's file (the same shape as moves' probe journal
428// on main; claim ask C14 confirms it as root authority). Every field root
429// uses that the claim also has is cross-checked against root's own claim
430// read: the op (the journal's header), the host, the record attempt k
431// (probe_attempts.<exec>), the holder's session, generation and stamp, and
432// the executor as the attempt's box. The read itself is readRelocation-
433// AttemptsBeneath's (I1).
434type relocationAttemptReader struct{ DataDir string }
435
436func (r relocationAttemptReader) ReadAttempt(ctx context.Context, o hostclaim.Observed) (hostclaim.WakeRetryAttempt, error) {
437 c := o.Claim
438 if r.DataDir == "" || c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRelocating || c.Op.Phase != hostclaim.PhaseRelocatingProbing || c.Op.Journal["stuck"] != "" {
439 return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
440 }
441 h := c.Holder
442 // P's opening keeps the install stamp; a later cold attempt runs at
443 // its own admission's stamp (generation.go: !KeepStamp sets stamp to
444 // the session), which the journaled start must name (Astra round 2
445 // finding 4). Main's claimLaunchAuthority still pins every relocating
446 // launch to install_stamp until claim ask C11.
447 if h.Generation == c.Op.Journal["prepared"] && c.Op.Journal["install_stamp"] != strconv.FormatUint(h.Stamp, 10) {
448 return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
449 }
450 k, err := strconv.Atoi(c.Op.Journal["probe_attempts."+c.Op.Exec])
451 if err != nil || k < 1 || k > hostclaim.MaxProbeAttempts {
452 return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
453 }
454 j, err := readRelocationAttemptsBeneath(ctx, r.DataDir, o.Host, c.Op.ID)
455 if err != nil {
456 return hostclaim.WakeRetryAttempt{}, err
457 }
458 if j.Sessions[strconv.FormatUint(h.Session, 10)] != k {
459 return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
460 }
461 st, ok := j.Starts[strconv.Itoa(k)]
462 if !ok || st.State != probeReserved || st.Session != h.Session || st.Generation != h.Generation || st.Stamp != h.Stamp || st.Attempt.Session != h.Session || st.Attempt.Box != c.Op.Exec {
463 return hostclaim.WakeRetryAttempt{}, hostbox.ErrChanged
464 }
465 return st.Attempt, nil
466}
467