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 "encoding/json"
6 "fmt"
7 "os/exec"
8 "strconv"
9 "sync"
10 "syscall"
11 "time"
12
13 "github.com/carlosframework/platform/internal/hostbox"
14 "github.com/carlosframework/platform/internal/hostclaim"
15 "github.com/carlosframework/platform/internal/hostwrite"
16 "github.com/carlosframework/platform/internal/moves"
17 "github.com/carlosframework/platform/internal/provision"
18 "github.com/carlosframework/platform/internal/release"
19)
20
21// movesProbeAttempts is the moves OperationAttempts implementation for a
22// landing or recovery probe (moves Phase 1 Plan 3 Task 5b; claim Plan 9 Group D, D2;
23// rulings H2, H10). The claim binds its methods into the admitted probe's
24// activator.ClaimStart; this type journals, counts and settles the probe's
25// starts in the box-local op journal (moves.Journal, <DataDir>/.moves/<host>/
26// <op>/), and writes no record and no claim.
27//
28// One record attempt is one admission. The landing (Task 6) counts it on the
29// record (provision.MoveProbeAttempt, Landing.Attempts = k) before it asks the
30// activator for the probe; every method here is bound to that k, refused
31// unless 1 <= k <= moves.AttemptsFor("landing", admitted), where admitted is
32// what T11 admitted for this move (Admitted, H10). Within an attempt the
33// activator makes up to five starts (startAdmitted).
34//
35// Steps it journals, per record attempt k:
36//
37// - probe_opened_<k>: the first app start's time, RFC 3339. OpenProbe
38// writes it once and returns it to every later start, a restarted
39// controller included, so the 30 s window is the attempt's and never
40// restarts.
41// - probe_start_<k>: the current start's reservation (probeStart): the
42// admitted session, the start's number n and the role launcher's
43// WakeRetryAttempt, and its state, reserved, failed or healthy.
44// - probe_session_<session>: the record attempt k the admitted session is
45// bound to, written by its first Reserve(0) or OpenProbe (fix round 1,
46// finding 5). A session serves one record attempt: once the record's
47// index moves past k every method refuses and CurrentAttempt answers
48// the zero attempt, so a new record attempt needs a new cold admission.
49//
50// Every method refuses (hostbox.ErrChanged) once the claim no longer names
51// the admitted writer (op id, kind, session, stamp, generation; phase
52// probing) or the record no longer names this landing at probing, unmarked,
53// at attempt k. A landing wake after the commit (Plan 3 Task 7) is served
54// too: its starts are journalled per admitted session (probe_start_committed-
55// <session>), bound to the claim at landing/committed and a committed record,
56// with no record attempt, no window and no settle.
57type movesProbeAttempts struct {
58 Claims release.Store
59 Records MovesRecordReader
60 DataDir string
61 // Admitted is the number of probe attempts T11 admitted for the move
62 // (H10: the target's report, or the record once MoveAdmit carries it).
63 Admitted func(context.Context, *provision.Record) (int, error)
64 NewID func() string
65 Now func() time.Time
66 // Settle is the wait before the second readiness request; zero is
67 // moves.SettleSeconds. Ready bounds that request; zero is
68 // moves.ProbeSeconds.
69 Settle, Ready time.Duration
70 // RecoveryAttempt, when set, is told of each recovery probe start this
71 // journal counts: the record, its recovery retry epoch and the start's
72 // number (the move log's probe_attempt-<epoch>-<n>, Plan 6 Task 3;
73 // movesWorker.recoveryAttemptLogger, set by movesWorker.probeAttempts).
74 RecoveryAttempt func(ctx context.Context, rec provision.Record, epoch uint64, n int)
75
76 mu sync.Mutex
77 current map[string]boundAttempt
78}
79
80// boundAttempt is a session's last reservation and the record attempt the
81// session is bound to.
82type boundAttempt struct {
83 k int
84 key string
85 attempt hostclaim.WakeRetryAttempt
86}
87
88var _ OperationAttempts = (*movesProbeAttempts)(nil)
89
90// Journal step names and start states.
91const (
92 probeStepOpened = "probe_opened_"
93 probeStepStart = "probe_start_"
94 probeStepSession = "probe_session_"
95
96 probeReserved = "reserved"
97 probeFailed = "failed"
98 probeHealthy = "healthy"
99)
100
101// probeStart is a probe_start_<k> step's value.
102type probeStart struct {
103 Session uint64 `json:"session"`
104 N int `json:"n"`
105 Attempt hostclaim.WakeRetryAttempt `json:"attempt"`
106 State string `json:"state"`
107}
108
109func (p *movesProbeAttempts) valid() bool {
110 return p != nil && p.Claims != nil && p.Records != nil && p.DataDir != "" && p.Admitted != nil && p.NewID != nil && p.Now != nil
111}
112
113// bound returns the landing record attempt, zero for a committed wake,
114// or -1 for a recovery probe. Every case validates the admitted writer.
115// Recovery also returns the budget key from the same record read.
116func (p *movesProbeAttempts) bound(ctx context.Context, s OperationStart) (int, string, error) {
117 k, key, _, err := p.boundRecord(ctx, s)
118 return k, key, err
119}
120
121// boundRecord is bound, with the record a recovery probe was bound against
122// (nil otherwise): Reserve logs a recovery start from it.
123func (p *movesProbeAttempts) boundRecord(ctx context.Context, s OperationStart) (int, string, *provision.Record, error) {
124 if !p.valid() {
125 return 0, "", nil, hostwrite.ErrUnavailable
126 }
127 a := s.Admitted
128 if a.Claim != nil && claimRecoveryKind(a.Claim.Op.Kind) {
129 key, rec, err := p.boundRecovery(ctx, s)
130 return -1, key, rec, err
131 }
132 k, key, err := p.boundLanding(ctx, s)
133 return k, key, nil, err
134}
135
136// boundLanding is bound for a landing's (or a committed wake's) probe.
137func (p *movesProbeAttempts) boundLanding(ctx context.Context, s OperationStart) (int, string, error) {
138 a := s.Admitted
139 if a.Claim == nil || a.Claim.Holder == nil || a.Claim.Op.Kind != hostclaim.KindLanding {
140 return 0, "", hostbox.ErrUnsupported
141 }
142 if !s.Policy.Probe {
143 return 0, "", p.boundCommitted(ctx, s)
144 }
145 h := a.Claim.Holder
146 if s.Owner.Host != a.Host || s.Owner.Session != h.Session || s.Owner.Stamp != h.Stamp || s.Owner.Incarnation != h.Incarnation {
147 return 0, "", hostbox.ErrChanged
148 }
149 fresh, err := hostclaim.Read(ctx, p.Claims, a.Host)
150 if err != nil {
151 return 0, "", err
152 }
153 if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseLandingProbing {
154 return 0, "", hostbox.ErrChanged
155 }
156 rec, err := p.Records(ctx, h.Account, h.App, a.Host)
157 if err != nil {
158 return 0, "", err
159 }
160 if rec == nil || rec.Host != a.Host || rec.Move == nil {
161 return 0, "", hostbox.ErrChanged
162 }
163 m := rec.Move
164 if m.ID != a.Claim.Op.ID || m.Incarnation != h.Incarnation || m.State != provision.MoveLanding || m.LandingPhase != provision.LandProbing || m.Reason != "" || m.Landing == nil {
165 return 0, "", hostbox.ErrChanged
166 }
167 admitted, err := p.Admitted(ctx, rec)
168 if err != nil {
169 return 0, "", err
170 }
171 k := m.Landing.Attempts
172 if k < 1 || k > moves.AttemptsFor(moves.ProbeLanding, admitted) {
173 return 0, "", hostbox.ErrChanged
174 }
175 return k, "", nil
176}
177
178// boundCommitted is bound for a committed wake: the claim still names the
179// admitted writer at landing/committed, and the record is this move,
180// committed. Its starts are journalled per session (startKey).
181func (p *movesProbeAttempts) boundCommitted(ctx context.Context, s OperationStart) error {
182 a := s.Admitted
183 h := a.Claim.Holder
184 if a.Claim.Op.Phase != hostclaim.PhaseLandingCommitted || s.Owner.Host != a.Host || s.Owner.Session != h.Session || s.Owner.Stamp != h.Stamp || s.Owner.Incarnation != h.Incarnation {
185 return hostbox.ErrChanged
186 }
187 fresh, err := hostclaim.Read(ctx, p.Claims, a.Host)
188 if err != nil {
189 return err
190 }
191 if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseLandingCommitted {
192 return hostbox.ErrChanged
193 }
194 rec, err := p.Records(ctx, h.Account, h.App, a.Host)
195 if err != nil {
196 return err
197 }
198 if rec == nil || rec.Host != a.Host || rec.Move == nil || rec.Move.ID != a.Claim.Op.ID || rec.Move.Incarnation != h.Incarnation || rec.Move.State != provision.MoveCommitted {
199 return hostbox.ErrChanged
200 }
201 return nil
202}
203
204// startKey names a start's journal step: the record attempt k for a probe,
205// the admitted session for a committed wake (k 0), or the recovery and
206// admitted session for a recovery probe (k -1).
207func startKey(s OperationStart, k int, recoveryKey string) string {
208 if k < 0 {
209 return recoveryKey + "-" + strconv.FormatUint(s.Owner.Session, 10)
210 }
211 if k == 0 {
212 return "committed-" + strconv.FormatUint(s.Owner.Session, 10)
213 }
214 return strconv.Itoa(k)
215}
216
217func (p *movesProbeAttempts) journal(s OperationStart) (*moves.Journal, error) {
218 return moves.ReadJournal(p.DataDir, s.Admitted.Host, s.Admitted.Claim.Op.ID)
219}
220
221func probeKey(s OperationStart) string {
222 return s.Admitted.Host + "\x00" + strconv.FormatUint(s.Owner.Session, 10)
223}
224
225// bindSession checks the session's journalled record attempt against k.
226// Absent, it is written into j when write is set (the caller writes j) and
227// refused otherwise.
228func bindSession(j *moves.Journal, s OperationStart, k int, write bool) error {
229 if k <= 0 {
230 return nil // committed wakes and recoveries have no landing record attempt
231 }
232 step := probeStepSession + strconv.FormatUint(s.Owner.Session, 10)
233 raw, ok := j.Steps[step]
234 switch {
235 case ok && raw == strconv.Itoa(k):
236 return nil
237 case ok:
238 return fmt.Errorf("%w: session %d is bound to record attempt %s, not %d", hostbox.ErrChanged, s.Owner.Session, raw, k)
239 case !write:
240 return fmt.Errorf("%w: session %d is bound to no record attempt", hostbox.ErrChanged, s.Owner.Session)
241 }
242 j.Mark(step, strconv.Itoa(k))
243 return nil
244}
245
246// OpenProbe journals the attempt's first start time, once.
247func (p *movesProbeAttempts) OpenProbe(ctx context.Context, s OperationStart, proposed time.Time) (time.Time, error) {
248 k, recoveryKey, err := p.bound(ctx, s)
249 if err != nil {
250 return time.Time{}, err
251 }
252 if k == 0 {
253 return time.Time{}, hostbox.ErrUnsupported
254 }
255 if proposed.IsZero() {
256 return time.Time{}, hostwrite.ErrUnavailable
257 }
258 p.mu.Lock()
259 defer p.mu.Unlock()
260 j, err := p.journal(s)
261 if err != nil {
262 return time.Time{}, err
263 }
264 if err := bindSession(j, s, k, true); err != nil {
265 return time.Time{}, err
266 }
267 step := probeStepOpened + startKey(s, k, recoveryKey)
268 if raw, ok := j.Steps[step]; ok {
269 if err := j.Commit(p.DataDir); err != nil {
270 return time.Time{}, err
271 }
272 opened, err := time.Parse(time.RFC3339Nano, raw)
273 if err != nil {
274 return time.Time{}, fmt.Errorf("moves journal %s/%s: %s: %w", s.Admitted.Host, s.Admitted.Claim.Op.ID, step, err)
275 }
276 return opened, nil
277 }
278 opened := proposed.UTC()
279 j.Mark(step, opened.Format(time.RFC3339Nano))
280 if err := j.Commit(p.DataDir); err != nil {
281 return time.Time{}, err
282 }
283 return opened, nil
284}
285
286// readStart returns attempt k's journalled start, or false.
287func readStart(j *moves.Journal, key string) (probeStart, bool, error) {
288 raw, ok := j.Steps[probeStepStart+key]
289 if !ok {
290 return probeStart{}, false, nil
291 }
292 var st probeStart
293 if err := json.Unmarshal([]byte(raw), &st); err != nil {
294 return probeStart{}, false, fmt.Errorf("moves journal %s/%s: %s%s: %w", j.Host, j.Op, probeStepStart, key, err)
295 }
296 return st, true, nil
297}
298
299func writeStart(dataDir string, j *moves.Journal, key string, st probeStart) error {
300 raw, err := json.Marshal(st)
301 if err != nil {
302 return err
303 }
304 j.Mark(probeStepStart+key, string(raw))
305 return j.Commit(dataDir)
306}
307
308// sameStartOwner reports whether a journalled reservation is the one this
309// admission's owner made for these launch inputs.
310func sameStartOwner(a hostclaim.WakeRetryAttempt, s OperationStart, digest string) bool {
311 o := s.Owner
312 return a.ID != "" && a.InvocationID != "" && a.Session == o.Session && a.Box == o.Box && a.Enrollment == o.Enrollment && a.BootID == o.BootID && a.ControllerID == o.ControllerID && a.LaunchInputs == digest
313}
314
315// Reserve journals start n of this admission before it runs. Start 0 is
316// reserved by the first Reserve(0) of the admission and validated by every
317// later one; start n > 0 only after start n-1 was settled failed.
318func (p *movesProbeAttempts) Reserve(ctx context.Context, s OperationStart, n int) (retErr error) {
319 k, recoveryKey, bindRec, err := p.boundRecord(ctx, s)
320 if err != nil {
321 return err
322 }
323 digest, err := ClaimLaunchDigest(s.Inputs)
324 if err != nil {
325 return err
326 }
327 p.mu.Lock()
328 defer p.mu.Unlock()
329 j, err := p.journal(s)
330 if err != nil {
331 return err
332 }
333 if err := bindSession(j, s, k, n == 0); err != nil {
334 return err
335 }
336 st, ok, err := readStart(j, startKey(s, k, recoveryKey))
337 if err != nil {
338 return err
339 }
340 same := ok && st.Session == s.Owner.Session
341 switch {
342 case n < 0:
343 return hostbox.ErrChanged
344 case n == 0 && same:
345 if st.N != 0 || st.State != probeReserved || !sameStartOwner(st.Attempt, s, digest) {
346 return hostbox.ErrChanged
347 }
348 case n == 0:
349 // The admission's first start; an earlier admission's journalled
350 // start, under another session, is over.
351 st = probeStart{Session: s.Owner.Session, N: 0, State: probeReserved}
352 default:
353 if !same || st.N != n-1 || st.State != probeFailed || st.Attempt.LaunchInputs != digest {
354 return hostbox.ErrChanged
355 }
356 st = probeStart{Session: s.Owner.Session, N: n, State: probeReserved}
357 }
358 if _, bound := j.Steps[probeStepSession+strconv.FormatUint(s.Owner.Session, 10)]; bound && st.Attempt.ID != "" {
359 // Reserve(0) validated an existing reservation; the binding it may
360 // have just written is made durable.
361 if err := j.Commit(p.DataDir); err != nil {
362 return err
363 }
364 }
365 if st.Attempt.ID == "" {
366 counted := 0
367 if k < 0 {
368 count, err := recoveryAttemptCount(j, recoveryKey)
369 if err != nil {
370 return err
371 }
372 if count >= recoveryProbeAttempts {
373 return ErrRecoveryProbeExhausted
374 }
375 counted = count + 1
376 j.Mark(recoveryKey, strconv.Itoa(counted))
377 }
378 defer func() {
379 if counted > 0 && retErr == nil {
380 p.recoveryAttempted(ctx, bindRec, counted)
381 }
382 }()
383 o := s.Owner
384 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}
385 if st.Attempt.ID == "" || st.Attempt.InvocationID == "" || st.Attempt.ID == st.Attempt.InvocationID {
386 return hostwrite.ErrUnavailable
387 }
388 if err := writeStart(p.DataDir, j, startKey(s, k, recoveryKey), st); err != nil {
389 return err
390 }
391 }
392 if p.current == nil {
393 p.current = map[string]boundAttempt{}
394 }
395 p.current[probeKey(s)] = boundAttempt{k: k, key: startKey(s, k, recoveryKey), attempt: st.Attempt}
396 return nil
397}
398
399// CurrentAttempt is the reservation the last Reserve journalled for this
400// admission; the zero attempt, which the role launcher refuses, before any.
401// It reads the record again (the interface gives it no context, so its own
402// bounded one): once the record's index is past the bound attempt it answers
403// the zero attempt.
404func (p *movesProbeAttempts) CurrentAttempt(s OperationStart) hostclaim.WakeRetryAttempt {
405 if p == nil {
406 return hostclaim.WakeRetryAttempt{}
407 }
408 p.mu.Lock()
409 cur, ok := p.current[probeKey(s)]
410 p.mu.Unlock()
411 if !ok {
412 return hostclaim.WakeRetryAttempt{}
413 }
414 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
415 defer cancel()
416 if k, recoveryKey, err := p.bound(ctx, s); err != nil || k != cur.k || startKey(s, k, recoveryKey) != cur.key {
417 return hostclaim.WakeRetryAttempt{}
418 }
419 return cur.attempt
420}
421
422// Failed settles start n, already stopped and drained, as failed.
423func (p *movesProbeAttempts) Failed(ctx context.Context, s OperationStart, n int) error {
424 recovery, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
425 defer cancel()
426 return p.settle(recovery, s, n, probeFailed)
427}
428
429func (p *movesProbeAttempts) settle(ctx context.Context, s OperationStart, n int, state string) error {
430 k, recoveryKey, err := p.bound(ctx, s)
431 if err != nil {
432 return err
433 }
434 p.mu.Lock()
435 defer p.mu.Unlock()
436 j, err := p.journal(s)
437 if err != nil {
438 return err
439 }
440 if err := bindSession(j, s, k, false); err != nil {
441 return err
442 }
443 st, ok, err := readStart(j, startKey(s, k, recoveryKey))
444 if err != nil {
445 return err
446 }
447 cur := p.current[probeKey(s)]
448 if !ok || st.Session != s.Owner.Session || st.N != n || st.State != probeReserved || cur.k != k || st.Attempt != cur.attempt {
449 return hostbox.ErrChanged
450 }
451 st.State = state
452 if err := writeStart(p.DataDir, j, startKey(s, k, recoveryKey), st); err != nil {
453 return err
454 }
455 if k < 0 && state == probeFailed {
456 count, err := recoveryAttemptCount(j, recoveryKey)
457 if err != nil {
458 return err
459 }
460 if count >= recoveryProbeAttempts {
461 return ErrRecoveryProbeExhausted
462 }
463 }
464 return nil
465}
466
467// Healthy accepts start n after readiness, replication and publication:
468// the 20 s settle with the process alive, then the probe's own readiness
469// check again (again, the claim's ClaimProbeReadiness against the started
470// role), with the process still alive after it (moves §7.7).
471func (p *movesProbeAttempts) Healthy(ctx context.Context, s OperationStart, n int, cmd *exec.Cmd, again func(context.Context) error) error {
472 k, _, err := p.bound(ctx, s)
473 if err != nil {
474 return err
475 }
476 p.mu.Lock()
477 j, err := p.journal(s)
478 if err == nil {
479 err = bindSession(j, s, k, false)
480 }
481 p.mu.Unlock()
482 if err != nil {
483 return err
484 }
485 if k == 0 {
486 // A committed wake: serving's readiness has passed, with
487 // replication and publication; there is no probe settle.
488 return p.settle(ctx, s, n, probeHealthy)
489 }
490 if again == nil || !probeAlive(cmd) {
491 return hostbox.ErrChanged
492 }
493 settle, ready := p.Settle, p.Ready
494 if settle <= 0 {
495 settle = time.Duration(moves.SettleSeconds * float64(time.Second))
496 }
497 if ready <= 0 {
498 ready = time.Duration(moves.ProbeSeconds * float64(time.Second))
499 }
500 timer := time.NewTimer(settle)
501 select {
502 case <-ctx.Done():
503 timer.Stop()
504 return ctx.Err()
505 case <-timer.C:
506 }
507 if !probeAlive(cmd) {
508 return fmt.Errorf("%w: the app exited during the settle", hostbox.ErrChanged)
509 }
510 asked, cancel := context.WithTimeout(ctx, ready)
511 err = again(asked)
512 cancel()
513 if err != nil {
514 return err
515 }
516 if !probeAlive(cmd) {
517 return fmt.Errorf("%w: the app exited after the second request", hostbox.ErrChanged)
518 }
519 return p.settle(ctx, s, n, probeHealthy)
520}
521
522// probeAlive reports whether cmd's process has not exited. Signal 0 checks
523// existence without delivering anything; once the activator's Wait has
524// reaped the process it reports os.ErrProcessDone.
525func probeAlive(cmd *exec.Cmd) bool {
526 return cmd != nil && cmd.Process != nil && cmd.Process.Signal(syscall.Signal(0)) == nil
527}
528
529// RecoveryProbeAttemptsReady is checked before the adapter records inputs.
530func (p *movesProbeAttempts) RecoveryProbeAttemptsReady() bool { return p.valid() }
531
532func (p *movesProbeAttempts) boundRecovery(ctx context.Context, s OperationStart) (string, *provision.Record, error) {
533 a := s.Admitted
534 if a.Claim.Holder == nil || !s.Policy.Probe {
535 return "", nil, hostbox.ErrChanged
536 }
537 h := a.Claim.Holder
538 if s.Owner.Host != a.Host || s.Owner.Incarnation != h.Incarnation || s.Owner.Stamp != h.Stamp || s.Owner.Session != h.Session || s.Owner.Box != a.Claim.Op.Exec {
539 return "", nil, hostbox.ErrChanged
540 }
541 fresh, err := hostclaim.Read(ctx, p.Claims, a.Host)
542 if err != nil {
543 return "", nil, err
544 }
545 if !claimSameWriter(a, fresh, hostclaim.SessionContinues) || fresh.Claim.Op.Phase != hostclaim.PhaseProbing || fresh.Claim.Op.Exec != s.Owner.Box {
546 return "", nil, hostbox.ErrChanged
547 }
548 rec, err := p.Records(ctx, h.Account, h.App, a.Host)
549 if err != nil {
550 return "", nil, err
551 }
552 if rec == nil || rec.Host != a.Host || rec.Move == nil || rec.Move.ID != a.Claim.Op.ID || rec.Move.Incarnation != h.Incarnation {
553 return "", nil, hostbox.ErrChanged
554 }
555 m := rec.Move
556 if a.Claim.Op.Kind == hostclaim.KindReverting {
557 if m.State != provision.MoveReverting || m.ReopenedAt != nil || fresh.Claim.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing {
558 return "", nil, hostbox.ErrChanged
559 }
560 } else if m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.ID == "" || m.Rollback.ID != a.Claim.Op.Intent["rollback"] || m.Rollback.ID != fresh.Claim.Op.Intent["rollback"] || m.Rollback.SourcePhase != provision.RbSourceProbing {
561 return "", nil, hostbox.ErrChanged
562 }
563 return recoveryAttemptKey(s, m.RecoveryRetryEpoch), rec, nil
564}
565
566// The move has its own journal directory. JSON preserves the rollback and
567// incarnation boundaries; sessions share this counter across controller restarts.
568func recoveryAttemptKey(s OperationStart, retryEpoch uint64) string {
569 c := s.Admitted.Claim
570 rollback := c.Op.ID
571 if c.Op.Kind == hostclaim.KindRestoring {
572 rollback = c.Op.Intent["rollback"]
573 }
574 parts := []string{c.Op.ID, rollback, c.Holder.Incarnation}
575 // Keep epoch zero's key so an upgrade preserves existing counts.
576 if retryEpoch != 0 {
577 parts = append(parts, strconv.FormatUint(retryEpoch, 10))
578 }
579 raw, _ := json.Marshal(parts)
580 return "recovery_attempts_" + string(raw)
581}
582
583func recoveryAttemptCount(j *moves.Journal, key string) (int, error) {
584 raw, ok := j.Steps[key]
585 if !ok {
586 return 0, nil
587 }
588 n, err := strconv.Atoi(raw)
589 if err != nil || n < 0 || n > recoveryProbeAttempts {
590 return 0, hostbox.ErrChanged
591 }
592 return n, nil
593}
594
595// recoveryAttempted tells RecoveryAttempt of start n of a recovery probe,
596// with the record the start was bound against (its retry epoch): the read
597// the binding already made, so a counted start is never lost to a second
598// read. The hook's own failed write goes to the outbox.
599func (p *movesProbeAttempts) recoveryAttempted(ctx context.Context, rec *provision.Record, n int) {
600 if p.RecoveryAttempt == nil || rec == nil || rec.Move == nil {
601 return
602 }
603 p.RecoveryAttempt(ctx, *rec, rec.Move.RecoveryRetryEpoch, n)
604}
605
606// probeAttempts is the attempt journal a box builds for its probe wakes
607// (Plan 3 Task 5), over the deployment store s, its records and newID, with
608// this worker's data directory and clock. Its RecoveryAttempt is the
609// worker's move log hook (recoveryAttemptLogger), so every recovery probe
610// start it counts is logged (fix round 2, ruling 2). Production's journal
611// is the registered wake source's (movesWakeSources.Open, Plan 5 Task 9b),
612// built the same way before this worker is armed; it reaches this worker's
613// hook through armedMovesWorker (movesWorkerRecoveryAttempt).
614func (w *movesWorker) probeAttempts(s release.Store, records MovesRecordReader, newID func() string) *movesProbeAttempts {
615 return &movesProbeAttempts{Claims: s, Records: records, DataDir: w.cfg.DataDir, Admitted: MovesRecordAdmittedAttempts,
616 NewID: newID, Now: w.cfg.Now, RecoveryAttempt: w.recoveryAttemptLogger(s)}
617}
618