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 "fmt"
7 "net"
8 "net/http"
9 "os"
10 "os/exec"
11 "sync"
12 "time"
13
14 "github.com/carlosframework/platform/internal/activator"
15 "github.com/carlosframework/platform/internal/cgroup"
16 "github.com/carlosframework/platform/internal/genselect"
17 "github.com/carlosframework/platform/internal/hostbox"
18 "github.com/carlosframework/platform/internal/hostclaim"
19 "github.com/carlosframework/platform/internal/hostguard"
20 "github.com/carlosframework/platform/internal/hostwrite"
21 "github.com/carlosframework/platform/internal/lease"
22 "github.com/carlosframework/platform/internal/registry"
23 "github.com/carlosframework/platform/internal/release"
24 "github.com/carlosframework/platform/internal/stopevidence"
25)
26
27// ClaimLaunchInputs is an in-process adapter result, never a new persisted
28// held-input format. The moves adapter retains its MoveInputs and manifest.
29type ClaimLaunchInputs struct {
30 Unit *hostbox.UnitLaunchInputs
31 claimWrite func(hostclaim.Observed, hostclaim.Observed) error
32 Roles []string
33 Identity hostwrite.Identity
34 Baseline *genselect.Plan
35 Admission hostclaim.Admission
36 Executable string
37 ExecutableSHA256 string
38 Args, Env []string
39 DBPath, Version string
40 // Staged is the file ExecutableSHA256 hashes and the version it is
41 // staged under (ClaimServingLaunch.Staged); Prepare hands it to the
42 // start, which records it before the exec (settings spec §4.2).
43 Staged release.StagedIdentity
44 Limit cgroup.Limit
45 Empty bool
46 Prepared bool
47 // ReadyPath is the probe's declared readiness path (moves §7.7,
48 // moves.ready_path). Serving and committed wakes ignore it.
49 ReadyPath string
50}
51
52type ClaimWakeInputs interface {
53 // Select is read-only. It verifies the exact ordinary or held input set.
54 Select(context.Context, registry.Route, hostclaim.Observed, activator.StartPolicy) (ClaimLaunchInputs, error)
55 // Materialize invokes captured RenderHeld/MaterializeHeld when held and
56 // establishes the origin-qualified files manifest before execution.
57 Materialize(context.Context, hostclaim.Observed, ClaimLaunchInputs, hostwrite.Work) error
58 EmptyAuthority(context.Context, hostclaim.Observed, ClaimLaunchInputs) error
59 Publish(context.Context, hostclaim.Observed, ClaimLaunchInputs) error
60}
61
62// ClaimPublication is what the assembly hands a fenced generation publisher
63// (claim §6.2a, moves §7.14 marker row; claim Plan 9 D5).
64type ClaimPublication struct {
65 // Sync is the session's current replicator's SyncAndWait: it ships what
66 // the app has committed and waits for the remote position. The publisher
67 // runs it before selecting the generation's root.
68 Sync func(context.Context) error
69 // Check is the publisher's fence: its producer is not drained, the
70 // attempt's lease is live and the captured selection still names the
71 // admitted session. It runs immediately before every marker request.
72 Check func(context.Context) error
73}
74
75// ClaimPublicationReceipt identifies an acknowledged marker write: the
76// version it created, or found on replay, and the exact root it named.
77type ClaimPublicationReceipt struct {
78 MarkerVersion string
79 Root genselect.File
80}
81
82// ErrClaimRootPending is a zero-position empty bootstrap before the app's
83// first transaction: the generation has no root, so no marker may name it.
84var ErrClaimRootPending = errors.New("host claims: the generation has no durable root yet")
85
86// ClaimFencedPublication writes generation.json for an admitted fresh
87// generation under a ClaimPublication and returns its receipt.
88type ClaimFencedPublication interface {
89 PublishFenced(context.Context, hostclaim.Observed, ClaimLaunchInputs, ClaimPublication) (ClaimPublicationReceipt, error)
90}
91
92// claimSyncer is the part of lease.Replicator the publisher uses.
93type claimSyncer interface {
94 SyncAndWait(context.Context) error
95}
96
97// ClaimWakeAssembly owns cold restore, lease transfer and failure cleanup.
98// Root supplies Stop and Launch; moves supplies only its input/publication API.
99type ClaimWakeAssembly struct {
100 seedFactory filesSeedRuntimeFactory
101 ServingPrerequisites func(context.Context, registry.Route, hostclaim.Observed) error
102 recoveryIntent func(context.Context, hostclaim.Observed, hostclaim.Admission) (hostclaim.Admission, error)
103 recoveryPrior func(context.Context, hostclaim.Observed) (hostclaim.Observed, bool, error)
104 recoveryPrelaunched func(context.Context, hostclaim.Observed) error
105 retained []*retainedClaimWork
106 AfterAdmission func(context.Context, hostclaim.Observed, hostclaim.Observed) error
107 BeforeAdmission func(context.Context, hostclaim.Observed) error
108 Roles *ClaimRoleAssembly
109 LeaseRegion string
110 Producers func(context.Context, hostclaim.Observed) (*ClaimProducers, error)
111 Inputs ClaimWakeInputs
112 Manager *cgroup.Manager
113 Retries activator.ServingRetries
114 Stop func(context.Context, hostclaim.Observed) error
115 RetireArtifacts func(context.Context, hostclaim.Observed, ClaimLaunchInputs, string) error
116 RecoverArtifacts func(context.Context, hostclaim.Observed, ClaimLaunchInputs) error
117 RequireServingEnrollment bool
118 BeforeLegacyInstall func(context.Context, hostclaim.Observed, ClaimLaunchInputs, hostwrite.Work) error
119 BeforeInstallLocked func(context.Context, hostclaim.Observed, ClaimLaunchInputs) error
120 // Prelaunch journals the admitted session's recoverable record before
121 // restore or materialization; Launch later completes that same record.
122 PrepareState func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs) error
123 Prelaunch func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs) (*activator.RoleLaunch, error)
124 Launch func(context.Context, hostclaim.Observed, hostwrite.Work, ClaimLaunchInputs, *activator.RoleLaunch) (*activator.RoleLaunch, error)
125 Ready func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs) error
126 Retain func(hostclaim.Observed, activator.Proc, activator.Rep, error)
127 ObserveLoss func(hostclaim.Observed, error)
128 // The moves C1 integration (claim Plan 9 Group D), nil unless both
129 // CARLOS_HOST_CLAIM and CARLOS_MOVES_PHASE1 are on. Continue is the
130 // session forward-step table (D1); nil keeps main's phase equality.
131 Continue func(kind, from, to string) bool
132 // Operations and ProbeReady admit and bind landing writers (D2, D3); nil
133 // keeps main's refusal of every probe and operation.
134 Operations OperationAttempts
135 ProbeReady ClaimProbeReadiness
136 // Publication is the fenced generation publisher (D5); nil keeps main's
137 // unfenced Inputs.Publish. Root binds it only with both flags on.
138 Publication ClaimFencedPublication
139 mu sync.Mutex
140 pending map[string]*claimWakeAttempt
141}
142type claimWakeAttempt struct {
143 resumedAdmission bool
144 releaseMu sync.Mutex
145 prelaunch *activator.RoleLaunch
146 launch *activator.RoleLaunch
147 repDone chan struct{}
148 repErr error
149 released bool
150 observed hostclaim.Observed
151 inputs ClaimLaunchInputs
152 held claimHeldLease
153 rep *lease.Replicator
154 syncer claimSyncer
155 // started is the attempt the latest operation start passed to Start.
156 started *hostclaim.WakeRetryAttempt
157 // publisher closes when the fenced publisher has joined; receipt and
158 // pubErr are its outcome.
159 publisher chan struct{}
160 receipt *ClaimPublicationReceipt
161 pubErr error
162 ctx context.Context
163 cancel context.CancelFunc
164 done chan struct{}
165 mu sync.Mutex
166 // owner is the assembly that admitted the attempt (StopWriters reads its
167 // producers).
168 owner *ClaimWakeAssembly
169 // The writers-only stop (claim Plan 11, R2). writersMu serialises
170 // StopWriters; writersClosed refuses new writer work once it began;
171 // writers is its witness once it succeeded. The journal and containment
172 // are the session's launch record's.
173 writersMu sync.Mutex
174 writersClosed bool
175 writers *ClaimWritersDrained
176 writersJournal claimWritersJournal
177 writersCgroup claimWritersCgroup
178 // replicatorFDs are the replicator's descriptors as this process saw
179 // them open (R3's agent half): root's narrowing hint, never evidence.
180 // replicatorFDsErr is why they could not be taken; root then exempts
181 // nothing and the observation fails closed.
182 replicatorFDs []stopevidence.DescriptorHint
183 replicatorFDsErr error
184}
185
186func (w *ClaimWakeAssembly) Build(ctx context.Context, source hostbox.WorkSource, writer hostbox.Writer) (*activator.ClaimRuntime, error) {
187 if w != nil && w.RequireServingEnrollment && (w.BeforeLegacyInstall == nil || w.BeforeInstallLocked == nil || w.RecoverArtifacts == nil) {
188 return nil, &ClaimDependencyError{Contracts: []string{"serving artifact enrollment and root retirement"}}
189 }
190 if w == nil || w.LeaseRegion == "" || w.AfterAdmission == nil || w.Inputs == nil || w.Manager == nil || w.Stop == nil || w.RetireArtifacts == nil || w.Retain == nil || w.ObserveLoss == nil || w.Producers == nil {
191 return nil, hostwrite.ErrUnavailable
192 }
193 if w.Launch == nil || w.Prelaunch == nil || w.PrepareState == nil {
194 if w.Roles == nil {
195 return nil, hostwrite.ErrUnavailable
196 }
197 if w.PrepareState == nil {
198 w.PrepareState = w.Roles.PrepareState
199 }
200 if w.Launch == nil {
201 w.Launch = w.Roles.Launch
202 }
203 if w.Prelaunch == nil {
204 w.Prelaunch = w.Roles.Prelaunch
205 }
206 }
207 if w.Retries.ExactHealth == nil {
208 w.Retries.ExactHealth = w.exactHealth
209 }
210 w.mu.Lock()
211 if w.pending != nil {
212 w.mu.Unlock()
213 return nil, hostwrite.ErrUnavailable
214 }
215 w.pending = map[string]*claimWakeAttempt{}
216 w.mu.Unlock()
217 if w.Ready == nil {
218 w.Ready = claimSocketReady
219 }
220 read := func(ctx context.Context, h string) (hostclaim.Observed, error) {
221 return hostclaim.Read(ctx, writer.Claims, h)
222 }
223 runtime := &activator.ClaimRuntime{Box: writer.Box, Read: read, RecoveryAttemptsReady: w.recoveryAttemptsReady}
224 runtime.ResumedAdmission = func(call context.Context, prior, admitted hostclaim.Observed) error {
225 a := w.get(admitted.Host)
226 if a == nil || !hostclaim.SameWriterIdentity(prior, admitted) {
227 return hostbox.ErrChanged
228 }
229 current, err := read(call, admitted.Host)
230 if err != nil {
231 return err
232 }
233 a.mu.Lock()
234 defer a.mu.Unlock()
235 if !a.resumedAdmission || !hostclaim.SameWriterIdentity(a.observed, admitted) || !hostclaim.SameWriterIdentity(current, admitted) {
236 return hostbox.ErrChanged
237 }
238 a.resumedAdmission = false
239 return nil
240 }
241 runtime.StopForeign = func(ctx context.Context, h string) error {
242 o, err := read(ctx, h)
243 if err != nil {
244 return err
245 }
246 return w.Stop(ctx, o)
247 }
248 runtime.Prerequisites = func(call context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) error {
249 if err := w.Roles.Preflight(call, ClaimLaunchInputs{}); err != nil {
250 return err
251 }
252 if o.Claim == nil {
253 return hostbox.ErrChanged
254 }
255 if err := w.Roles.PreflightKind(call, o.Claim.Op.Kind); err != nil {
256 return err
257 }
258 if err := w.admitsOperation(o, p, writer.Box); err != nil {
259 return err
260 }
261 if rt.Seed != "" || rt.IsSidecar() || rt.Tunnel != "" {
262 return hostbox.ErrUnsupported
263 }
264 prior, _, err := w.pendingRecovery(call, o)
265 if err != nil {
266 return err
267 }
268 in, err := w.Inputs.Select(call, rt, prior, p)
269 if err != nil {
270 return err
271 }
272 if (rt.Unit != "" || rt.Backing != "") && in.Unit == nil {
273 return hostbox.ErrUnsupported
274 }
275 if len(in.Roles) != 1 || in.Roles[0] != "main" || in.Executable == "" || len(in.ExecutableSHA256) != 64 || in.DBPath == "" || in.Identity.Config == "" || in.Identity.Release == "" || in.Admission.Mode != "cold" || (in.Baseline == nil && !in.Empty) {
276 return hostbox.ErrUnsupported
277 }
278 if o.Claim == nil || o.Claim.Holder == nil {
279 return hostbox.ErrChanged
280 }
281 if err := w.Roles.PreflightLaunch(rt.Host, in); err != nil {
282 return err
283 }
284 // The predecessor must be stopped before another lease or session opens.
285 if err := w.stopPredecessor(call, o); err != nil {
286 return err
287 }
288 bounded, cancel := context.WithTimeout(call, hostbox.RequestLimit)
289 held, err := lease.Acquire(bounded, writer.Bucket, o.Claim.Holder.Location, w.LeaseRegion)
290 cancel()
291 if err != nil {
292 return err
293 }
294 live, stop := context.WithCancel(ctx)
295 a := &claimWakeAttempt{owner: w, observed: o, inputs: in, held: held, ctx: live, cancel: stop, done: make(chan struct{})}
296 w.mu.Lock()
297 previous := w.pending[rt.Host]
298 if previous == nil {
299 w.pending[rt.Host] = a
300 }
301 w.mu.Unlock()
302 if previous != nil {
303 stop()
304 return errors.Join(hostbox.ErrChanged, held.Release(call))
305 }
306 go func() {
307 select {
308 case <-held.Lost():
309 stop()
310 a.mu.Lock()
311 current := a.observed
312 a.mu.Unlock()
313 close(a.done)
314 cleanup, cancel := context.WithTimeout(context.Background(), 30*time.Second)
315 err := w.stopAttempt(cleanup, a, current)
316 cancel()
317 w.ObserveLoss(current, errors.Join(a.leaseErr(), err))
318 case <-live.Done():
319 close(a.done)
320 }
321 }()
322 return nil
323 }
324 runtime.Admit = func(call context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (result hostclaim.Observed, resultErr error) {
325 a := w.get(rt.Host)
326 if a == nil {
327 return hostclaim.Observed{}, hostwrite.ErrUnavailable
328 }
329 // ServingRetries owns serving cleanup. Operations bypass that wrapper,
330 // so an admission error must release their pending lease here. No
331 // Prepare or app execution can have run before Admit returns. Keep the
332 // claim and all durable reservations intact, including an uncertain CAS.
333 if w.operationKind(o) {
334 defer func() {
335 if resultErr == nil {
336 return
337 }
338 cleanup, cancel := context.WithTimeout(context.Background(), hostbox.RequestLimit)
339 defer cancel()
340 current, _ := a.snapshot()
341 stopErr := w.stopAttempt(cleanup, a, current)
342 resultErr = errors.Join(resultErr, stopErr, w.release(cleanup, rt.Host, a))
343 }()
344 }
345 if err := a.leaseErr(); err != nil {
346 return hostclaim.Observed{}, err
347 }
348 // The retry reservation changed the ETag. Re-select the typed provider
349 // against that observation; never re-use authorization from the old read.
350 prior, resuming, err := w.pendingRecovery(call, o)
351 if err != nil {
352 return hostclaim.Observed{}, err
353 }
354 in, err := w.Inputs.Select(call, rt, prior, p)
355 if err != nil {
356 return hostclaim.Observed{}, err
357 }
358 admitted := o
359 if !resuming {
360 if w.recoveryIntent != nil {
361 in.Admission, err = w.recoveryIntent(call, o, in.Admission)
362 if err != nil {
363 return admitted, err
364 }
365 }
366 admitted, err = writer.Admit(call, o, in.Admission, in.Baseline)
367 } else {
368 in.Admission.NewGeneration = o.Claim.Holder.Generation
369 }
370 if err != nil {
371 return admitted, err
372 }
373 a.mu.Lock()
374 a.observed, a.inputs = admitted, in
375 a.mu.Unlock()
376 if err := w.AfterAdmission(call, prior, admitted); err != nil {
377 return admitted, err
378 }
379 if err := source.Guard.Activate(call, rt.Host, writer.Box, nil); err != nil {
380 return admitted, err
381 }
382 // The launch record precedes the retry bind, restore and materialization.
383 // Any later failure or crash leaves the exact tuple root stops and settles.
384 if err := w.persistPrelaunch(call, a, source, admitted, in); err != nil {
385 return admitted, err
386 }
387 if w.recoveryPrelaunched != nil {
388 if err := w.recoveryPrelaunched(call, admitted); err != nil {
389 return admitted, err
390 }
391 }
392 a.mu.Lock()
393 a.resumedAdmission = resuming
394 a.mu.Unlock()
395 return admitted, nil
396 }
397 runtime.Prepare = func(call context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (activator.ClaimStart, error) {
398 a := w.get(rt.Host)
399 if a == nil {
400 return activator.ClaimStart{}, hostbox.ErrChanged
401 }
402 if err := a.admitsWriters(); err != nil {
403 return activator.ClaimStart{}, err
404 }
405 if !a.advance(o, w.Continue) {
406 return activator.ClaimStart{}, hostbox.ErrChanged
407 }
408 a.mu.Lock()
409 pre := a.prelaunch
410 a.mu.Unlock()
411 // No restore or materialization runs without the pre-launch record.
412 if pre == nil || pre.Target.Host != o.Host || pre.Target.Incarnation != o.Claim.Holder.Incarnation || pre.Target.Stamp != o.Claim.Holder.Stamp || pre.Target.Session != o.Claim.Holder.Session {
413 return activator.ClaimStart{}, hostbox.ErrChanged
414 }
415 selected, err := source.SelectAt(call, rt.Host, claimSpawnAction(o.Claim), hostguard.Captured{Incarnation: o.Claim.Holder.Incarnation, Stamp: o.Claim.Holder.Stamp})
416 if err != nil {
417 return activator.ClaimStart{}, err
418 }
419 _, in := a.snapshot()
420 observation := &claimRestoreObservation{observed: o, box: writer.Box, dataDir: source.Guard.DataDir, db: in.DBPath}
421 in.claimWrite = observation.retag
422 selected = claimLeaseWork{Work: selected, Alive: a.ctx, Check: a.leaseErr}
423 restore := hostbox.ColdRestore{Reader: writer.Data, Bucket: writer.Bucket, Generation: o.Claim.Holder.Generation, Work: selected, Identity: in.Identity, RetireArtifacts: func(ctx context.Context, g string) error { return w.RetireArtifacts(ctx, o, in, g) }}
424 if o.Claim.Op.Kind == hostclaim.KindServing {
425 if w.BeforeLegacyInstall != nil {
426 restore.BeforeLegacyInstall = func(ctx context.Context, path, generation string) error {
427 if path != in.DBPath || generation != o.Claim.Holder.Generation {
428 return hostbox.ErrChanged
429 }
430 return w.BeforeLegacyInstall(ctx, o, in, selected)
431 }
432 }
433 if w.BeforeInstallLocked != nil {
434 restore.BeforeInstallLocked = func(ctx context.Context, path, generation string) error {
435 if path != in.DBPath || generation != o.Claim.Holder.Generation {
436 return hostbox.ErrChanged
437 }
438 return w.BeforeInstallLocked(ctx, o, in)
439 }
440 }
441 }
442 s := activator.ClaimStart{Live: a.ctx, Work: selected, Identity: in.Identity, Manager: w.Manager, Host: rt.Host, Limit: in.Limit, Version: in.Version, Staged: in.Staged}
443 s.Restore = func(call context.Context) error {
444 if err := a.admitsWriters(); err != nil {
445 return err
446 }
447 if w.RecoverArtifacts != nil {
448 if err := w.RecoverArtifacts(call, o, in); err != nil {
449 return err
450 }
451 }
452 if err := w.PrepareState(call, o, selected, in); err != nil {
453 return err
454 }
455 if err := a.leaseErr(); err != nil {
456 return err
457 }
458 if in.Empty {
459 return restore.InstallEmpty(call, in.DBPath, func(ctx context.Context) error { return w.Inputs.EmptyAuthority(ctx, o, in) })
460 }
461 return restore.Install(call, in.DBPath, in.Baseline)
462 }
463 s.RetryClaimWrite = observation.retry
464 s.Materialize = func(ctx context.Context) error {
465 expected := observation.snapshot()
466 fresh, err := read(ctx, o.Host)
467 if err != nil {
468 return err
469 }
470 if !sameClaimObservation(expected, fresh) {
471 return hostclaim.ErrConflict
472 }
473 if !a.advance(expected, w.Continue) {
474 return hostbox.ErrChanged
475 }
476 return w.Inputs.Materialize(ctx, expected, in, selected)
477 }
478
479 s.Build = func() *exec.Cmd {
480 cmd := exec.Command(in.Executable, in.Args...)
481 cmd.Env = append([]string(nil), in.Env...)
482 return cmd
483 }
484 ready, err := w.readiness(o, p, rt, in, a, pre)
485 if err != nil {
486 return activator.ClaimStart{}, err
487 }
488 s.Ready = ready
489 s.PrepareLaunch = func(ctx context.Context) (*activator.RoleLaunch, error) {
490 if err := a.admitsWriters(); err != nil {
491 return nil, err
492 }
493 launch, err := w.Launch(ctx, o, selected, in, pre)
494 if err != nil {
495 return nil, err
496 }
497 if launch == nil {
498 return nil, hostwrite.ErrUnavailable
499 }
500 producers, err := w.Producers(ctx, o)
501 if err != nil {
502 return nil, err
503 }
504 launch.RegisterProducer = func(call context.Context, id string) (activator.RoleProducer, error) {
505 if err := call.Err(); err != nil {
506 return activator.RoleProducer{}, err
507 }
508 producerCtx, finish, err := producers.Register(a.ctx, launch.Target, id)
509 return activator.RoleProducer{Context: producerCtx, Finish: finish}, err
510 }
511 a.mu.Lock()
512 a.launch = launch
513 a.mu.Unlock()
514 a.bindWriters(launch)
515 return launch, nil
516 }
517 s.Replicate = func(call context.Context, prefix string) (activator.Rep, error) {
518 // Prepared generation markers precede replication; ordinary markers
519 // follow the first durable root through s.Publish below.
520 if in.Prepared {
521 if err := w.Inputs.Publish(call, o, in); err != nil {
522 return nil, err
523 }
524 }
525 check := func(ctx context.Context) error { return selected.With(ctx, in.Identity, func() error { return nil }) }
526 if err := a.held.BindGeneration(prefix, check); err != nil {
527 return nil, err
528 }
529 producers, err := w.Producers(call, o)
530 if err != nil {
531 return nil, err
532 }
533 producerCtx, finished, err := producers.Register(a.ctx, a.launch.Target, ClaimProducerReplicator)
534 if err != nil {
535 return nil, err
536 }
537 var rep *lease.Replicator
538 hints, hintErr := recordReplicatorHints(call, "/proc", os.Getpid(), in.DBPath, func() error {
539 var err error
540 rep, err = a.held.Replicate(producerCtx, in.DBPath, 0)
541 return err
542 })
543 if rep == nil {
544 finished(hintErr)
545 return nil, hintErr
546 }
547 a.mu.Lock()
548 a.replicatorFDs, a.replicatorFDsErr = hints, hintErr
549 a.rep, a.syncer = rep, rep
550 a.repDone = make(chan struct{})
551 a.mu.Unlock()
552 go func() {
553 <-producerCtx.Done()
554 cleanup, cancel := context.WithTimeout(context.Background(), hostbox.RequestLimit)
555 err := rep.Close(cleanup)
556 cancel()
557 a.mu.Lock()
558 a.repErr = err
559 a.mu.Unlock()
560 finished(err)
561 close(a.repDone)
562 }()
563 return &claimOwnedRep{attempt: a, owner: w, host: rt.Host, rep: rep}, nil
564 }
565 s.Publish = func(ctx context.Context, current hostclaim.Observed) (hostclaim.Observed, error) {
566 if !in.Prepared && w.Publication != nil {
567 return w.publish(ctx, a, o, current, in, selected, read, writer.Published)
568 }
569 if !in.Prepared {
570 if err := w.Inputs.Publish(ctx, current, in); err != nil {
571 return hostclaim.Observed{}, err
572 }
573 }
574 return writer.Published(ctx, current)
575 }
576 s.Retain = func(p activator.Proc, r activator.Rep, err error) { w.Retain(o, p, r, err) }
577 if o.Claim.Op.Kind != hostclaim.KindServing {
578 if err := w.bindOperation(&s, OperationStart{Admitted: o, Policy: p, Inputs: in, Owner: pre.Target}); err != nil {
579 return activator.ClaimStart{}, err
580 }
581 // Probe readiness finds the role cgroup by the exact attempt this
582 // start passed to Start, never by another CurrentAttempt read.
583 current := s.CurrentAttempt
584 s.CurrentAttempt = func() hostclaim.WakeRetryAttempt {
585 attempt := current()
586 a.mu.Lock()
587 a.started = &attempt
588 a.mu.Unlock()
589 return attempt
590 }
591 }
592 return s, nil
593 }
594 runtime.FailedAdmission = func(ctx context.Context, o hostclaim.Observed, cause error) error {
595 a := w.get(o.Host)
596 if a == nil {
597 return nil
598 }
599 current, _ := a.snapshot()
600 if err := w.stopAttempt(ctx, a, current); err != nil {
601 return err
602 }
603 return w.release(ctx, o.Host, a)
604 }
605 wrapped, err := w.Retries.Wrap(runtime)
606 if err != nil {
607 return nil, err
608 }
609 if w.ServingPrerequisites != nil {
610 previous := wrapped.Prerequisites
611 wrapped.Prerequisites = func(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) error {
612 if !p.Probe {
613 if err := w.ServingPrerequisites(ctx, rt, o); err != nil {
614 return err
615 }
616 }
617 return previous(ctx, rt, o, p)
618 }
619 }
620 return wrapped, nil
621}
622
623// claimHeldLease is the part of lease.Held one admitted attempt uses.
624type claimHeldLease interface {
625 Err() error
626 Lost() <-chan struct{}
627 Release(context.Context) error
628 BindGeneration(string, func(context.Context) error) error
629 Replicate(context.Context, string, time.Duration) (*lease.Replicator, error)
630}
631
632// persistPrelaunch records the admitted session under the captured guard. It
633// runs once per attempt and binds exactly the admitted writer session.
634func (w *ClaimWakeAssembly) persistPrelaunch(ctx context.Context, a *claimWakeAttempt, source hostbox.WorkSource, admitted hostclaim.Observed, in ClaimLaunchInputs) error {
635 if w.Prelaunch == nil || admitted.Claim == nil || admitted.Claim.Holder == nil {
636 return hostwrite.ErrUnavailable
637 }
638 a.mu.Lock()
639 existing := a.prelaunch
640 a.mu.Unlock()
641 if existing != nil {
642 return hostbox.ErrChanged
643 }
644 h := admitted.Claim.Holder
645 selected, err := source.SelectAt(ctx, admitted.Host, claimSpawnAction(admitted.Claim), hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp})
646 if err != nil {
647 return err
648 }
649 pre, err := w.Prelaunch(ctx, admitted, claimLeaseWork{Work: selected, Alive: a.ctx, Check: a.leaseErr}, in)
650 if err != nil {
651 return err
652 }
653 if pre == nil || pre.Journal == nil || pre.Target.Host != admitted.Host || pre.Target.Incarnation != h.Incarnation || pre.Target.Stamp != h.Stamp || pre.Target.Session != h.Session {
654 if pre != nil && pre.Parent != nil {
655 return errors.Join(hostbox.ErrChanged, pre.Parent.Close())
656 }
657 return hostbox.ErrChanged
658 }
659 a.mu.Lock()
660 a.prelaunch = pre
661 a.mu.Unlock()
662 a.bindWriters(pre)
663 return nil
664}
665
666// bindWriters records the launch record's journal and containment for
667// StopWriters. A record without either leaves the binding unchanged.
668func (a *claimWakeAttempt) bindWriters(l *activator.RoleLaunch) {
669 if l == nil || l.Journal == nil || l.Parent == nil {
670 return
671 }
672 a.mu.Lock()
673 a.writersJournal, a.writersCgroup = l.Journal, l.Parent
674 a.mu.Unlock()
675}
676
677// publish runs a fresh generation's marker publication as a producer of the
678// session's exact launch (claim §6.2a): root's fence cancels it and joins it
679// before acknowledging the drain, and its check runs before every marker
680// request. The publisher syncs the current replicator before it selects the
681// root. A zero-position empty bootstrap has no root until the app's first
682// transaction (R8-3, moves §7.14 "the writer waits"): the same producer
683// keeps waiting for it in the background, and the claim records publication
684// only from its receipt.
685func (w *ClaimWakeAssembly) publish(ctx context.Context, a *claimWakeAttempt, admitted, current hostclaim.Observed, in ClaimLaunchInputs, selected hostwrite.Work, read func(context.Context, string) (hostclaim.Observed, error), published func(context.Context, hostclaim.Observed) (hostclaim.Observed, error)) (hostclaim.Observed, error) {
686 a.mu.Lock()
687 launch, syncer, previous := a.launch, a.syncer, a.publisher
688 a.mu.Unlock()
689 if w.Publication == nil || launch == nil || syncer == nil || selected == nil || read == nil || published == nil {
690 return hostclaim.Observed{}, hostwrite.ErrUnavailable
691 }
692 if previous != nil || !a.advance(current, w.Continue) {
693 return hostclaim.Observed{}, hostbox.ErrChanged
694 }
695 producers, err := w.Producers(ctx, admitted)
696 if err != nil {
697 return hostclaim.Observed{}, err
698 }
699 producerCtx, finish, err := producers.Register(a.ctx, launch.Target, ClaimProducerPublisher)
700 if err != nil {
701 return hostclaim.Observed{}, err
702 }
703 done := make(chan struct{})
704 a.mu.Lock()
705 a.publisher = done
706 a.mu.Unlock()
707 // Every request has returned when the publisher settles, so it is
708 // joined; its outcome is the receipt, not drain evidence.
709 settle := func(receipt ClaimPublicationReceipt, err error) {
710 a.mu.Lock()
711 if err == nil {
712 a.receipt = &receipt
713 }
714 a.pubErr = err
715 a.mu.Unlock()
716 close(done)
717 finish(nil)
718 }
719 pub := ClaimPublication{Sync: syncer.SyncAndWait, Check: func(ctx context.Context) error {
720 if err := producerCtx.Err(); err != nil {
721 return err
722 }
723 return selected.With(ctx, in.Identity, producerCtx.Err)
724 }}
725 joined, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
726 stop := context.AfterFunc(producerCtx, cancel)
727 receipt, err := w.Publication.PublishFenced(joined, current, in, pub)
728 stop()
729 cancel()
730 if in.Empty && errors.Is(err, ErrClaimRootPending) {
731 go func() {
732 settle(w.awaitRoot(producerCtx, a, admitted, in, pub, read, published))
733 }()
734 return current, nil
735 }
736 settle(receipt, err)
737 if err != nil {
738 return hostclaim.Observed{}, err
739 }
740 return published(ctx, current)
741}
742
743// awaitRoot is the zero-position publisher's wait for the first durable
744// root, with a doubling backoff from one second to thirty while the producer
745// lives. Each try reads the claim afresh and is bounded like the first. The
746// fence, a changed writer or a refused marker ends it without a receipt.
747func (w *ClaimWakeAssembly) awaitRoot(ctx context.Context, a *claimWakeAttempt, admitted hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication, read func(context.Context, string) (hostclaim.Observed, error), published func(context.Context, hostclaim.Observed) (hostclaim.Observed, error)) (ClaimPublicationReceipt, error) {
748 delay := time.Second
749 for {
750 timer := time.NewTimer(delay)
751 select {
752 case <-ctx.Done():
753 timer.Stop()
754 return ClaimPublicationReceipt{}, ctx.Err()
755 case <-timer.C:
756 }
757 delay = min(2*delay, 30*time.Second)
758 current, err := read(ctx, admitted.Host)
759 if err == nil && !a.advance(current, w.Continue) {
760 return ClaimPublicationReceipt{}, hostbox.ErrChanged
761 }
762 if err != nil {
763 continue
764 }
765 bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit)
766 receipt, err := w.Publication.PublishFenced(bounded, current, in, pub)
767 cancel()
768 switch {
769 case errors.Is(err, ErrClaimRootPending):
770 continue
771 case errors.Is(err, hostbox.ErrChanged) || ctx.Err() != nil:
772 return ClaimPublicationReceipt{}, errors.Join(err, ctx.Err())
773 case err != nil:
774 continue
775 }
776 // A failed claim CAS retries; the marker replay finds this receipt.
777 if _, err := published(ctx, current); err == nil {
778 return receipt, nil
779 }
780 }
781}
782
783func (w *ClaimWakeAssembly) get(host string) *claimWakeAttempt {
784 w.mu.Lock()
785 defer w.mu.Unlock()
786 return w.pending[host]
787}
788func (w *ClaimWakeAssembly) release(ctx context.Context, host string, a *claimWakeAttempt) error {
789 a.releaseMu.Lock()
790 defer a.releaseMu.Unlock()
791 a.cancel()
792 select {
793 case <-a.done:
794 case <-ctx.Done():
795 return ctx.Err()
796 }
797 a.mu.Lock()
798 publisher := a.publisher
799 a.mu.Unlock()
800 // The publisher's producer context is the attempt's: it is cancelled
801 // above, and its last request returns before the lease is released.
802 if publisher != nil {
803 select {
804 case <-publisher:
805 case <-ctx.Done():
806 return ctx.Err()
807 }
808 }
809 a.mu.Lock()
810 done := a.repDone
811 released := a.released
812 previous := a.repErr
813 a.mu.Unlock()
814 if released {
815 return previous
816 }
817 var err error
818 if done != nil {
819 select {
820 case <-done:
821 case <-ctx.Done():
822 return ctx.Err()
823 }
824 a.mu.Lock()
825 err = a.repErr
826 a.mu.Unlock()
827 } else {
828 err = a.held.Release(ctx)
829 }
830 if err == nil {
831 a.mu.Lock()
832 parent := (*cgroup.Prepared)(nil)
833 if a.launch != nil {
834 parent = a.launch.Parent
835 } else if a.prelaunch != nil {
836 parent = a.prelaunch.Parent
837 }
838 a.mu.Unlock()
839 if parent != nil {
840 err = parent.Close()
841 }
842 }
843 a.mu.Lock()
844 a.released = true
845 a.repErr = err
846 a.mu.Unlock()
847 if err == nil {
848 w.mu.Lock()
849 if w.pending[host] == a {
850 delete(w.pending, host)
851 }
852 w.mu.Unlock()
853 }
854 return err
855}
856func (w *ClaimWakeAssembly) Close(ctx context.Context) error {
857 w.mu.Lock()
858 pending := make(map[string]*claimWakeAttempt, len(w.pending))
859 for h, a := range w.pending {
860 pending[h] = a
861 }
862 w.mu.Unlock()
863 var errs []error
864 for h, a := range pending {
865 current, _ := a.snapshot()
866 if err := w.stopAttempt(ctx, a, current); err != nil {
867 errs = append(errs, err)
868 continue
869 }
870 errs = append(errs, w.release(ctx, h, a))
871 }
872 return errors.Join(errs...)
873}
874
875type claimOwnedRep struct {
876 attempt *claimWakeAttempt
877 owner *ClaimWakeAssembly
878 host string
879 rep *lease.Replicator
880}
881
882func (r *claimOwnedRep) Err() error { return r.rep.Err() }
883
884// Close is the activator's stop of the replica. Once StopWriters has begun,
885// the replicator belongs to the writers-stopped release: root's stop drains
886// it with the other producers and Wake.Stop then releases the lease, so the
887// activator's close leaves it running (claim Plan 11, R2).
888func (r *claimOwnedRep) Close(ctx context.Context) error {
889 r.attempt.mu.Lock()
890 stopped := r.attempt.writersClosed
891 r.attempt.mu.Unlock()
892 if stopped {
893 return nil
894 }
895 return r.owner.release(ctx, r.host, r.attempt)
896}
897
898// claimWakeContinue is the wake and selected-work checkpoint rule: an admitted
899// writer may observe a later phase of its own row after missing one (m-2). The
900// artifact retirement and the claim's own transitions keep SessionContinues.
901var claimWakeContinue = hostclaim.SessionObserves
902
903// claimSameWriter reports whether b, observed at or after a, still names a's
904// admitted writer session. continues is the session forward-step table; nil
905// requires the same phase, as main did.
906func claimSameWriter(a, b hostclaim.Observed, continues func(kind, from, to string) bool) bool {
907 if a.Claim == nil || b.Claim == nil || a.Claim.Holder == nil || b.Claim.Holder == nil {
908 return false
909 }
910 x, y := a.Claim, b.Claim
911 phase := x.Op.Phase == y.Op.Phase
912 if continues != nil {
913 phase = continues(x.Op.Kind, x.Op.Phase, y.Op.Phase)
914 }
915 return a.Host == b.Host && x.Holder.Incarnation == y.Holder.Incarnation && x.Holder.Stamp == y.Holder.Stamp && x.Holder.Session == y.Holder.Session && x.Holder.Generation == y.Holder.Generation && x.Op.ID == y.Op.ID && phase && x.Op.Kind == y.Op.Kind && x.Op.Exec == y.Op.Exec
916}
917
918// admitsOperation is the probe and operation admission path (claim Plan 9
919// D2, claim Plan 11 R1). Ordinary serving wakes keep main's path unchanged.
920// With the moves C1 owners bound, the K6 table admits a landing probe while
921// the op is at probing and a landing wake once it is committed; with region
922// HA's relocation owners also bound, it admits a relocation target's probe
923// at probing where op.exec is this box, and no ordinary relocating wake.
924// Every other probe or operation keeps main's typed refusal, before any
925// lease, stop or reservation.
926func (w *ClaimWakeAssembly) admitsOperation(o hostclaim.Observed, p activator.StartPolicy, box string) error {
927 if !p.Probe && (o.Claim == nil || o.Claim.Op.Kind == hostclaim.KindServing) {
928 return nil
929 }
930 ops, ready := w.operationOwner(o)
931 if ops == nil || (p.Probe && ready == nil) || (!p.Probe && o.Claim != nil && claimRecoveryKind(o.Claim.Op.Kind)) {
932 return claimOperationRefusal()
933 }
934 return hostbox.WakePhase(o.Claim, box, p.Probe)
935}
936
937// bindOperation binds the moves attempt journal to one admitted operation
938// writer, as ServingRetries does for serving. The closures carry the exact
939// admitted observation; moves refuses once it no longer holds.
940func (w *ClaimWakeAssembly) bindOperation(s *activator.ClaimStart, start OperationStart) error {
941 ops, _ := w.operationOwner(start.Admitted)
942 if ops == nil || s == nil || s.Ready == nil {
943 return claimOperationRefusal()
944 }
945 ready := s.Ready
946 if start.Policy.Probe {
947 s.OpenProbe = func(ctx context.Context, proposed time.Time) (time.Time, error) {
948 return ops.OpenProbe(ctx, start, proposed)
949 }
950 }
951 s.CurrentAttempt = func() hostclaim.WakeRetryAttempt { return ops.CurrentAttempt(start) }
952 s.Reserve = func(ctx context.Context, n int) error { return ops.Reserve(ctx, start, n) }
953 s.Failed = func(ctx context.Context, n int) error { return ops.Failed(ctx, start, n) }
954 s.Healthy = func(ctx context.Context, n int, cmd *exec.Cmd) error {
955 return ops.Healthy(ctx, start, n, cmd, func(ctx context.Context) error { return ready(ctx, cmd) })
956 }
957 return nil
958}
959
960// readiness selects the readiness check by op kind (claim Plan 9 D3).
961// Serving and a landing wake after the commit keep main's check: GET /, any
962// status below 500, the exact spawned pid. A landing probe uses the probe
963// check against the started main role's cgroup, which the role's journalled
964// reservation names: the exact attempt the start passed to Start.
965func (w *ClaimWakeAssembly) readiness(o hostclaim.Observed, p activator.StartPolicy, rt registry.Route, in ClaimLaunchInputs, a *claimWakeAttempt, pre *activator.RoleLaunch) (func(context.Context, *exec.Cmd) error, error) {
966 if !p.Probe {
967 return func(ctx context.Context, cmd *exec.Cmd) error {
968 a.mu.Lock()
969 launch := a.launch
970 a.mu.Unlock()
971 if launch != nil && launch.Root != nil {
972 if in.Unit != nil {
973 return nil
974 }
975 return claimPollReady(ctx, rt.Socket, "http://carlos/", func(conn net.Conn) error { return launch.PeerInRole(ctx, conn, cmd) }, func(status int) bool { return status < 500 })
976 }
977 return w.Ready(ctx, cmd, rt, in)
978 }, nil
979 }
980 ops, probeReady := w.operationOwner(o)
981 if ops == nil || probeReady == nil || pre == nil {
982 return nil, claimOperationRefusal()
983 }
984 return func(ctx context.Context, cmd *exec.Cmd) error {
985 a.mu.Lock()
986 launch, started := a.launch, a.started
987 a.mu.Unlock()
988 if launch == nil || started == nil {
989 return hostwrite.ErrUnavailable
990 }
991 group, err := launch.RoleCgroup(ctx, cmd, *started)
992 if err != nil {
993 return err
994 }
995 return probeReady(ctx, cmd, rt, in, group)
996 }, nil
997}
998
999func (a *claimWakeAttempt) snapshot() (hostclaim.Observed, ClaimLaunchInputs) {
1000 a.mu.Lock()
1001 defer a.mu.Unlock()
1002 return a.observed, a.inputs
1003}
1004
1005// advance is the attempt's session checkpoint (M9): next replaces the last
1006// checked observation only if it still names the same writer one adjacent
1007// forward step on at most (claimSameWriter under continues). A skipped or
1008// backward step refuses and leaves the checkpoint as it was.
1009func (a *claimWakeAttempt) advance(next hostclaim.Observed, continues func(kind, from, to string) bool) bool {
1010 a.mu.Lock()
1011 defer a.mu.Unlock()
1012 if !claimSameWriter(a.observed, next, continues) {
1013 return false
1014 }
1015 a.observed = next
1016 return true
1017}
1018func (a *claimWakeAttempt) leaseErr() error {
1019 if err := a.ctx.Err(); err != nil {
1020 return err
1021 }
1022 a.mu.Lock()
1023 rep := a.rep
1024 a.mu.Unlock()
1025 if rep != nil {
1026 return rep.Err()
1027 }
1028 return a.held.Err()
1029}
1030
1031type claimLeaseWork struct {
1032 Work hostwrite.Work
1033 Alive context.Context
1034 Check func() error
1035}
1036
1037func (w claimLeaseWork) With(ctx context.Context, id hostwrite.Identity, fn func() error) error {
1038 if w.Work == nil || w.Alive == nil || w.Check == nil || fn == nil {
1039 return hostwrite.ErrUnavailable
1040 }
1041 joined, cancel := context.WithCancel(ctx)
1042 defer cancel()
1043 stop := context.AfterFunc(w.Alive, cancel)
1044 defer stop()
1045 return w.Work.With(joined, id, func() error {
1046 if err := w.Alive.Err(); err != nil {
1047 return err
1048 }
1049 if err := joined.Err(); err != nil {
1050 return err
1051 }
1052 if err := w.Check(); err != nil {
1053 return err
1054 }
1055 return fn()
1056 })
1057}
1058func claimSocketReady(ctx context.Context, cmd *exec.Cmd, rt registry.Route, _ ClaimLaunchInputs) error {
1059 if cmd == nil || cmd.Process == nil || rt.Socket == "" {
1060 return hostwrite.ErrUnavailable
1061 }
1062 pid := cmd.Process.Pid
1063 return claimPollReady(ctx, rt.Socket, "http://carlos/", func(conn net.Conn) error { return claimReadyPeer(conn, pid) }, func(status int) bool { return status < 500 })
1064}
1065
1066// claimPollReady polls target over the role's socket until accept takes the
1067// response status. Every connection's peer is verified before a request is
1068// written on it; redirects are answers, never followed.
1069func claimPollReady(ctx context.Context, socket, target string, peer func(net.Conn) error, accept func(int) bool) error {
1070 transport := &http.Transport{Proxy: nil, DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
1071 conn, err := (&net.Dialer{}).DialContext(ctx, "unix", socket)
1072 if err != nil {
1073 return nil, err
1074 }
1075 if err = peer(conn); err != nil {
1076 conn.Close()
1077 return nil, err
1078 }
1079 return conn, nil
1080 }, DisableKeepAlives: true}
1081 defer transport.CloseIdleConnections()
1082 client := &http.Client{Transport: transport, Timeout: time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
1083 timer := time.NewTicker(150 * time.Millisecond)
1084 defer timer.Stop()
1085 for {
1086 if err := ctx.Err(); err != nil {
1087 return err
1088 }
1089 req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
1090 if err != nil {
1091 return err
1092 }
1093 resp, err := client.Do(req)
1094 if err == nil {
1095 resp.Body.Close()
1096 if accept(resp.StatusCode) {
1097 return nil
1098 }
1099 err = fmt.Errorf("readiness HTTP %d", resp.StatusCode)
1100 }
1101 select {
1102 case <-ctx.Done():
1103 return errors.Join(ctx.Err(), err)
1104 case <-timer.C:
1105 }
1106 }
1107}
1108
1109func (w *ClaimWakeAssembly) exactHealth(ctx context.Context, cmd *exec.Cmd, attempt hostclaim.WakeRetryAttempt) error {
1110 w.mu.Lock()
1111 pending := make([]*claimWakeAttempt, 0, len(w.pending))
1112 for _, a := range w.pending {
1113 pending = append(pending, a)
1114 }
1115 w.mu.Unlock()
1116 for _, a := range pending {
1117 a.mu.Lock()
1118 launch := a.launch
1119 a.mu.Unlock()
1120 if launch != nil {
1121 if err := launch.ExactHealth(ctx, cmd, attempt); err == nil {
1122 return nil
1123 }
1124 }
1125 }
1126 return hostbox.ErrChanged
1127}
1128
1129// stopPredecessor uses sealed cross-box evidence only when production wired
1130// that path. The ordinary local stop remains the default.
1131func (w *ClaimWakeAssembly) pendingRecovery(ctx context.Context, o hostclaim.Observed) (hostclaim.Observed, bool, error) {
1132 if w.recoveryPrior == nil {
1133 return o, false, nil
1134 }
1135 return w.recoveryPrior(ctx, o)
1136}
1137func (w *ClaimWakeAssembly) stopPredecessor(ctx context.Context, o hostclaim.Observed) error {
1138 prior, pending, err := w.pendingRecovery(ctx, o)
1139 if err != nil {
1140 return err
1141 }
1142 if pending {
1143 return w.AfterAdmission(ctx, prior, o)
1144 }
1145 if w.BeforeAdmission != nil {
1146 return w.BeforeAdmission(ctx, o)
1147 }
1148 return w.Stop(ctx, o)
1149}
1150
1151// stopAttempt never sends a predecessor on another box to the local root.
1152// An operation that has not made a prelaunch record cannot have launched any
1153// roles or started replication. Releasing that unused lease is not a stop
1154// receipt or a refund: the claim, effect and attempt journals remain intact.
1155func (w *ClaimWakeAssembly) stopAttempt(ctx context.Context, a *claimWakeAttempt, current hostclaim.Observed) error {
1156 a.mu.Lock()
1157 pre := a.prelaunch
1158 a.mu.Unlock()
1159 if w.Continue != nil && pre == nil && w.operationKind(current) {
1160 return nil
1161 }
1162 // Session zero admitted no writer; the predecessor witness already ran.
1163 if pre == nil && current.Claim != nil && current.Claim.Holder != nil && current.Claim.Holder.Session == 0 {
1164 return nil
1165 }
1166 return w.Stop(ctx, current)
1167}
1168
1169// claimRecoveryKind identifies the source recovery operations.
1170func claimRecoveryKind(kind string) bool {
1171 return kind == hostclaim.KindReverting || kind == hostclaim.KindRestoring
1172}
1173