| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "log" |
| 7 | "sync" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/hostclaim" |
| 11 | "github.com/carlosframework/platform/internal/hostops" |
| 12 | "github.com/carlosframework/platform/internal/hostwrite" |
| 13 | ) |
| 14 | |
| 15 | // The owning agent drains its own producers over the existing authenticated |
| 16 | // root socket. It never receives root's evidence key or issues a stop receipt. |
| 17 | func (a *ClaimAssembly) startWakeResetWorker(ctx context.Context) error { |
| 18 | if a.Mode != "agent" { |
| 19 | return nil |
| 20 | } |
| 21 | if a.RetirementAuthority == nil || a.Close == nil || a.Wake == nil || a.Launches == nil { |
| 22 | return hostwrite.ErrUnavailable |
| 23 | } |
| 24 | actor, err := a.Identity.Verify(ctx) |
| 25 | if err != nil { |
| 26 | return err |
| 27 | } |
| 28 | transport := hostops.WakeResetTransport{Store: a.Claims, Directory: a.RetirementAuthority.Directory, Now: time.Now} |
| 29 | worker, cancel := context.WithCancel(ctx) |
| 30 | var wg sync.WaitGroup |
| 31 | wg.Add(1) |
| 32 | go func() { |
| 33 | defer wg.Done() |
| 34 | poll := resetPollMin |
| 35 | for worker.Err() == nil { |
| 36 | listed, err := transport.Scan(worker, actor.Box, a.stopResetAttempt) |
| 37 | if err != nil && worker.Err() == nil { |
| 38 | log.Printf("host claim reset: %v", err) |
| 39 | } |
| 40 | poll = nextResetPoll(poll, listed) |
| 41 | timer := time.NewTimer(poll) |
| 42 | select { |
| 43 | case <-worker.Done(): |
| 44 | timer.Stop() |
| 45 | return |
| 46 | case <-timer.C: |
| 47 | } |
| 48 | } |
| 49 | }() |
| 50 | closeOwners := a.Close |
| 51 | a.Close = func(ctx context.Context) error { cancel(); wg.Wait(); return closeOwners(ctx) } |
| 52 | return nil |
| 53 | } |
| 54 | |
| 55 | // The worker lists its box's reset prefix once a second while requests are |
| 56 | // present and backs off to resetPollMax while idle, so a box with no resets |
| 57 | // does not LIST the bucket every second. |
| 58 | const ( |
| 59 | resetPollMin = time.Second |
| 60 | resetPollMax = 8 * time.Second |
| 61 | ) |
| 62 | |
| 63 | func nextResetPoll(prev time.Duration, listed int) time.Duration { |
| 64 | if listed > 0 || prev < resetPollMin { |
| 65 | return resetPollMin |
| 66 | } |
| 67 | return min(2*prev, resetPollMax) |
| 68 | } |
| 69 | |
| 70 | func (a ClaimAssembly) stopResetAttempt(ctx context.Context, o hostclaim.Observed, attempt hostclaim.WakeRetryAttempt) error { |
| 71 | if o.Claim == nil || o.Claim.Holder == nil || o.Claim.Holder.WakeRetry == nil || o.Claim.Holder.WakeRetry.Attempt == nil || *o.Claim.Holder.WakeRetry.Attempt != attempt { |
| 72 | return hostwrite.ErrUnavailable |
| 73 | } |
| 74 | local, err := a.localStopIdentity(ctx) |
| 75 | if err != nil { |
| 76 | return err |
| 77 | } |
| 78 | if local.Actor.Box != attempt.Box || local.Actor.Enrollment != attempt.Enrollment { |
| 79 | return hostwrite.ErrUnavailable |
| 80 | } |
| 81 | h := o.Claim.Holder |
| 82 | entries, err := a.Launches.SessionLaunches(ctx, o.Host, h.Incarnation, h.Stamp, h.Session, attempt.Box, attempt.Enrollment) |
| 83 | if err != nil { |
| 84 | return err |
| 85 | } |
| 86 | for _, entry := range entries { |
| 87 | owner := ownerOf(entry) |
| 88 | if owner.BootID == attempt.BootID && owner.ControllerID == attempt.ControllerID { |
| 89 | _, err = a.collectCapturedStop(ctx, o, local, owner, false) |
| 90 | return err |
| 91 | } |
| 92 | } |
| 93 | return errors.Join(hostops.ErrWakeResetUncertain, ErrClaimStopInventory) |
| 94 | } |
| 95 | |
amadan