| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/activator" |
| 10 | "github.com/carlosframework/platform/internal/hostbox" |
| 11 | "github.com/carlosframework/platform/internal/hostclaim" |
| 12 | "github.com/carlosframework/platform/internal/moves" |
| 13 | "github.com/carlosframework/platform/internal/provision" |
| 14 | "github.com/carlosframework/platform/internal/registry" |
| 15 | "github.com/carlosframework/platform/internal/release" |
| 16 | ) |
| 17 | |
| 18 | // The source's recovery probe adapter (moves §7.8 source probing; claim |
| 19 | // Plan 10 J2). It is the production recoveryProbe hook: movesrecover.go's |
| 20 | // probing calls it once the record says probing and the claim is this |
| 21 | // rollback's restoring op at probing with this box executing. |
| 22 | // |
| 23 | // One call is one probe-shaped claim wake, in this order: |
| 24 | // |
| 25 | // 1. the request is checked against the record (this rollback at probing, |
| 26 | // nine attempts, the coverage and generation of its return; a declared |
| 27 | // return, whose record cites the target's declared loss or fence, asks |
| 28 | // no coverage, Q6) and the claim is read fresh (restoring/probing, this |
| 29 | // op, this rollback, this box executing, the record's incarnation, a |
| 30 | // clean return's src_gen, the target's loss the record cites); |
| 31 | // 2. the inputs the probe will render are selected (the box's current |
| 32 | // inputs, as the cut reads them) and recorded through req.Record, which |
| 33 | // is provision.MoveRecordRecoveryInputs under the move and rollback ids, |
| 34 | // legal only at probing. A clean return records here: it has no |
| 35 | // configured phase. A dirty return records again, replacing what its |
| 36 | // install recorded at configured, because this render is the probe's. |
| 37 | // A failed record write stops the attempt: nothing is rendered or run; |
| 38 | // 3. the activator is asked for a probe-shaped claim wake (StartPolicy |
| 39 | // Probe, carrying exactly the recorded inputs). That wake is the claim's |
| 40 | // Group D path unchanged: Prerequisites and admitsOperation admit a |
| 41 | // recovery probe only at probing with the moves C1 owners bound; Admit, |
| 42 | // the pre-launch record and Prepare bind OperationAttempts and the |
| 43 | // role-cgroup probe readiness to the admitted session; the moves Inputs |
| 44 | // owner selects and materializes (the held render) the recorded inputs, |
| 45 | // and moves' policy admits only those (MovesClaimPolicy). |
| 46 | // |
| 47 | // The adapter writes nothing besides req.Record. The restore, install and |
| 48 | // evidence remain the moves owners'; the nine-attempt budget is journalled |
| 49 | // by moves' OperationAttempts, which refuses a start past it with |
| 50 | // ErrRecoveryProbeExhausted. |
| 51 | |
| 52 | // ClaimRecoveryProbeWake is the activator's recovery probe entry: one |
| 53 | // claim wake of rt under p (p.Probe set), through ClaimRuntime's |
| 54 | // Prerequisites, Admit and Prepare and the activator's own start loop. It |
| 55 | // returns nil once a start was healthy (OperationAttempts.Healthy accepted |
| 56 | // it). Production binds Service.RecoveryProbe, which retains the process |
| 57 | // in the activator and refuses missing claim-runtime dependencies. |
| 58 | type ClaimRecoveryProbeWake func(ctx context.Context, rt registry.Route, p activator.StartPolicy) error |
| 59 | |
| 60 | // ErrRecoveryProbeExhausted is the moves attempt journal's refusal once the |
| 61 | // recovery's nine starts are spent. The recovery stops for an operator at |
| 62 | // probing (StuckRecoveryProbeAttempts). |
| 63 | var ErrRecoveryProbeExhausted = errors.New("moves: the recovery probe's attempts are spent") |
| 64 | |
| 65 | // errRecoveryProbeRequest is a request that does not match its record: a |
| 66 | // caller bug, never retried into a wake. |
| 67 | var errRecoveryProbeRequest = errors.New("moves: the recovery probe request does not match its record") |
| 68 | |
| 69 | // errRecoveryInputsNoRelease: the selected inputs name no release, which the |
| 70 | // record transition would refuse; the adapter refuses first. |
| 71 | var errRecoveryInputsNoRelease = errors.New("moves: the recovery probe's inputs name no release") |
| 72 | |
| 73 | // recoveryProbeStartsPerWake is the starts one probe wake may run. The |
| 74 | // activator allows at most five per wake; the recovery's nine are a budget |
| 75 | // across wakes, journalled by moves. |
| 76 | const recoveryProbeStartsPerWake = 3 |
| 77 | |
| 78 | // recoveryProbeWindow is the probe's readiness window from its first |
| 79 | // app start (§7.7's 30 s). |
| 80 | const recoveryProbeWindow = 30 * time.Second |
| 81 | |
| 82 | // recoveryProbeAdapter is the source's recovery probe. Claims is the |
| 83 | // captured CARLOS_HOST_CLAIM; Route reads this box's registry row; Current |
| 84 | // selects the inputs the probe will render; Wake is the activator's entry. |
| 85 | type recoveryProbeAdapter struct { |
| 86 | Box string |
| 87 | Claims bool |
| 88 | Route func(host string) (*registry.Route, error) |
| 89 | Current func(ctx context.Context, s release.Store, rec provision.Record) (provision.MoveInputs, error) |
| 90 | Wake ClaimRecoveryProbeWake |
| 91 | Ready func() bool |
| 92 | } |
| 93 | |
| 94 | // armRecoveryProbe binds the production recoveryProbe hook: the adapter |
| 95 | // over this worker's configuration, read when the probe runs. wake is the |
| 96 | // activator's recovery probe entry; tests may leave it nil to prove refusal. |
| 97 | func armRecoveryProbe(cfg *moveCfg, w *movesWorker, wake ClaimRecoveryProbeWake, ready func() bool) { |
| 98 | cfg.recoveryProbe = func(ctx context.Context, s release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) { |
| 99 | c := w.cfg |
| 100 | a := recoveryProbeAdapter{Box: c.Box, Claims: c.claimRecords, Route: c.Route, Current: w.recoveryInputs, Wake: wake, Ready: ready} |
| 101 | return a.probe(ctx, s, rec, req) |
| 102 | } |
| 103 | } |
| 104 | |
| 105 | // probe runs one recovery probe wake for rec (see the file comment). |
| 106 | func (a recoveryProbeAdapter) probe(ctx context.Context, s release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) { |
| 107 | var missing []string |
| 108 | if !a.Claims { |
| 109 | missing = append(missing, "CARLOS_HOST_CLAIM Group D recovery admission") |
| 110 | } |
| 111 | if a.Wake == nil || (a.Ready != nil && !a.Ready()) { |
| 112 | missing = append(missing, "activator recovery probe entry") |
| 113 | } |
| 114 | if a.Box == "" || a.Route == nil || a.Current == nil { |
| 115 | missing = append(missing, "moves recovery probe route and inputs") |
| 116 | } |
| 117 | if len(missing) > 0 { |
| 118 | return recoveryProbeWaiting, &ClaimDependencyError{Contracts: missing} |
| 119 | } |
| 120 | if err := recoveryProbeMatches(rec, req); err != nil { |
| 121 | return recoveryProbeWaiting, err |
| 122 | } |
| 123 | obs, err := hostclaim.Read(ctx, s, rec.Host) |
| 124 | if err != nil { |
| 125 | return recoveryProbeWaiting, fmt.Errorf("read the host claim: %w", err) |
| 126 | } |
| 127 | if err := a.admits(obs, rec, req); err != nil { |
| 128 | return recoveryProbeWaiting, err |
| 129 | } |
| 130 | rt, err := a.Route(rec.Host) |
| 131 | if err != nil { |
| 132 | return recoveryProbeWaiting, fmt.Errorf("read the registry row: %w", err) |
| 133 | } |
| 134 | if rt == nil || rt.Host != rec.Host { |
| 135 | return recoveryProbeWaiting, fmt.Errorf("%w: %s has no registry row on this box", hostbox.ErrChanged, rec.Host) |
| 136 | } |
| 137 | in, err := a.Current(ctx, s, rec) |
| 138 | if err != nil { |
| 139 | return recoveryProbeWaiting, fmt.Errorf("select the recovery inputs: %w", err) |
| 140 | } |
| 141 | if in.Release == "" { |
| 142 | return recoveryProbeWaiting, errRecoveryInputsNoRelease |
| 143 | } |
| 144 | // The record precedes every render and launch of this wake. |
| 145 | if err := req.Record(ctx, in); err != nil { |
| 146 | return recoveryProbeWaiting, fmt.Errorf("record the recovery inputs: %w", err) |
| 147 | } |
| 148 | recorded := in |
| 149 | recorded.BundleETags = make(map[string]string, len(in.BundleETags)) |
| 150 | for k, v := range in.BundleETags { |
| 151 | recorded.BundleETags[k] = v |
| 152 | } |
| 153 | err = a.Wake(ctx, *rt, activator.StartPolicy{RecoveryFence: moveFenceGeneration(rec.Move), MaxStarts: recoveryProbeStartsPerWake, Window: recoveryProbeWindow, Probe: true, Inputs: &recorded}) |
| 154 | switch { |
| 155 | case err == nil: |
| 156 | return recoveryProbeAnswered, nil |
| 157 | case errors.Is(err, ErrRecoveryProbeExhausted): |
| 158 | return recoveryProbeExhausted, nil |
| 159 | case errors.Is(err, moves.ErrAttemptOver), errors.Is(err, hostbox.ErrChanged): |
| 160 | // This wake's window lapsed or its admission no longer holds: the |
| 161 | // next pass probes again, and only the probe. |
| 162 | return recoveryProbeWaiting, nil |
| 163 | } |
| 164 | return recoveryProbeWaiting, fmt.Errorf("the recovery probe wake: %w", err) |
| 165 | } |
| 166 | |
| 167 | // recoveryProbeMatches checks the request against its record: this |
| 168 | // rollback, rolling back with the source at probing, the nine-attempt |
| 169 | // budget and the coverage and generation of its return. |
| 170 | func recoveryProbeMatches(rec provision.Record, req recoveryProbeRequest) error { |
| 171 | m := rec.Move |
| 172 | if m != nil && m.ID != "" && m.State == provision.MoveReverting { |
| 173 | if m.ReopenedAt != nil || req.RollbackID != "" || req.Coverage != "" || req.Generation != "" || !req.Clean || req.Declared || req.Attempts != recoveryProbeAttempts || req.Record == nil { |
| 174 | return errRecoveryProbeRequest |
| 175 | } |
| 176 | return nil |
| 177 | } |
| 178 | if m == nil || m.ID == "" || m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.ID != req.RollbackID || m.Rollback.SourcePhase != provision.RbSourceProbing { |
| 179 | return fmt.Errorf("%w: not rollback %s at probing", errRecoveryProbeRequest, req.RollbackID) |
| 180 | } |
| 181 | if req.Record == nil || req.Attempts != recoveryProbeAttempts || req.Clean == m.Rollback.Dirty { |
| 182 | return fmt.Errorf("%w: budget %d, clean %v", errRecoveryProbeRequest, req.Attempts, req.Clean) |
| 183 | } |
| 184 | // The declared kind (Q6): a return whose record cites a declared loss |
| 185 | // or a fence of the target is clean and asks no coverage. |
| 186 | declared := m.Rollback.LossRef != "" |
| 187 | switch { |
| 188 | case req.Declared != declared: |
| 189 | return fmt.Errorf("%w: the request's declared return is not the record's", errRecoveryProbeRequest) |
| 190 | // SourceGeneration may be "" (whole-branch fixes, item 4): the claim's |
| 191 | // legacy encoding of a source whose sessions predate admission records, |
| 192 | // which recoverRun.probing passes through from the journal and the |
| 193 | // owners normalise as hostclaim.RootGeneration; admits holds it to the |
| 194 | // claim's. |
| 195 | case declared && (!req.Clean || req.Coverage != "" || req.Generation != ""): |
| 196 | return fmt.Errorf("%w: a declared return admits a fresh generation restored from src_gen, uncovered", errRecoveryProbeRequest) |
| 197 | case declared: |
| 198 | case req.Clean: |
| 199 | if req.Coverage != recoveryCoverageClean || req.Generation != "" { |
| 200 | return fmt.Errorf("%w: a clean return admits a fresh generation restored from src_gen", errRecoveryProbeRequest) |
| 201 | } |
| 202 | case req.Coverage != recoveryCoverageSealed || req.Generation != req.RollbackID: |
| 203 | return fmt.Errorf("%w: a dirty return admits its sealed return generation", errRecoveryProbeRequest) |
| 204 | } |
| 205 | return nil |
| 206 | } |
| 207 | |
| 208 | // admits checks the fresh claim: this move's restoring op for this |
| 209 | // rollback, at probing, executed by this box under the record's |
| 210 | // incarnation, naming the request's src_gen. Anything else waits. |
| 211 | func (a recoveryProbeAdapter) admits(obs hostclaim.Observed, rec provision.Record, req recoveryProbeRequest) error { |
| 212 | c := obs.Claim |
| 213 | if obs.Absent || c == nil || c.Holder == nil || c.Op.ID != rec.Move.ID || c.Op.Phase != hostclaim.PhaseProbing || c.Op.Exec != a.Box || c.Holder.Incarnation != rec.Move.Incarnation { |
| 214 | return hostbox.ErrChanged |
| 215 | } |
| 216 | if rec.Move.State == provision.MoveReverting { |
| 217 | if c.Op.Kind != hostclaim.KindReverting || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing || c.Holder.Generation != req.SourceGeneration { |
| 218 | return hostbox.ErrChanged |
| 219 | } |
| 220 | return nil |
| 221 | } |
| 222 | if c.Op.Kind != hostclaim.KindRestoring || c.Op.Intent["rollback"] != req.RollbackID { |
| 223 | return hostbox.ErrChanged |
| 224 | } |
| 225 | // The fresh claim must still agree with the record on the target's |
| 226 | // loss, and with the request's kind (Q6). |
| 227 | declared, err := recoveryDeclared(c, rec.Move.Rollback.LossRef) |
| 228 | if err != nil { |
| 229 | return err |
| 230 | } |
| 231 | if declared != req.Declared { |
| 232 | return fmt.Errorf("%w: the claim's loss of the target is not the request's", hostbox.ErrChanged) |
| 233 | } |
| 234 | // The claim's src_gen and the request's, normalised as the owners |
| 235 | // normalise them (selectedGeneration: "" is the legacy root), must agree. |
| 236 | if selectedGeneration(c.Op.Journal[hostclaim.JournalSrcGen]) != selectedGeneration(req.SourceGeneration) { |
| 237 | return fmt.Errorf("%w: the claim's src_gen is not the request's", hostbox.ErrChanged) |
| 238 | } |
| 239 | return nil |
| 240 | } |
| 241 | |
amadan