| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/hostclaim" |
| 10 | execctl "github.com/carlosframework/platform/internal/hostexec" |
| 11 | "github.com/carlosframework/platform/internal/hostguard" |
| 12 | "github.com/carlosframework/platform/internal/hostrelocate" |
| 13 | "github.com/carlosframework/platform/internal/release" |
| 14 | ) |
| 15 | |
| 16 | // RelocationIdentity is this box's verified actor and stop identity, as |
| 17 | // Plan 6b's assembly reports it: main's claimStopLocal, which |
| 18 | // localStopIdentity fills (actor, boot, controller, and this process's PID |
| 19 | // and start time). Release records the process with the flushed positions |
| 20 | // and decides whether a recorded flush is this process's own with main's |
| 21 | // claimLaunchOwner.local rule. |
| 22 | type RelocationIdentity = claimStopLocal |
| 23 | |
| 24 | // ReleasePosition is one recorded release: the exact stop request, the |
| 25 | // process that flushed it (PID and start time, beside the request's boot |
| 26 | // and controller), and the positions its flush reached. |
| 27 | type ReleasePosition struct { |
| 28 | Request StopRequest `json:"request"` |
| 29 | PID int `json:"pid"` |
| 30 | ProcessStart string `json:"process_start"` |
| 31 | TXID uint64 `json:"txid"` |
| 32 | Seq int64 `json:"seq"` |
| 33 | } |
| 34 | |
| 35 | // owner is the recorded flush's launch tuple in main's form, so the |
| 36 | // decision whether it is this process's own is claimLaunchOwner.local's. |
| 37 | func (p ReleasePosition) owner() claimLaunchOwner { |
| 38 | return claimLaunchOwner{BootID: p.Request.BootID, ControllerID: p.Request.ControllerID, DirectoryRevision: p.Request.ExpectedDirectoryRevision, |
| 39 | PID: p.PID, ProcessStart: p.ProcessStart} |
| 40 | } |
| 41 | |
| 42 | // ReleasePositions keeps the flushed positions of one release durably, |
| 43 | // bound to the exact stop request (host, incarnation, stamp, session, |
| 44 | // operation, phase, box, enrollment, boot, controller) and the flushing |
| 45 | // process, from before root completes its receipt. A replayed receipt (root's durable fence after a |
| 46 | // lost response or a crash before JournalRelease) is paired with them; |
| 47 | // without a record for that exact request the release is refused rather |
| 48 | // than guessed. |
| 49 | // Recorded matches the stop's installation and session (host, incarnation, |
| 50 | // stamp, session, operation, phase, obligation, box, enrollment), not its |
| 51 | // controller identity: a restarted agent asks with a new boot and |
| 52 | // controller id (or the same ids in a new process), and gets back the |
| 53 | // request the flush was made under, so it can ask root for that stop's |
| 54 | // durable receipt. |
| 55 | type ReleasePositions interface { |
| 56 | Record(ctx context.Context, p ReleasePosition) error |
| 57 | Recorded(ctx context.Context, r StopRequest) (recorded ReleasePosition, ok bool, err error) |
| 58 | } |
| 59 | |
| 60 | // RelocationEffects is hostrelocate.Effects over providers the claim |
| 61 | // assembly supplies (rev 2). It owns only the order of §5.8's release |
| 62 | // sequence and the shape of each record; every provider acts under the |
| 63 | // host guard. The execution controller is main's: one participation owner |
| 64 | // per box (ruling 42), opened by productionClaimLocal for every non-serving |
| 65 | // operation that lists this box, whichever of the claim's participation |
| 66 | // loop, WorkSource.Prepare or this pass asks first. |
| 67 | type RelocationEffects struct { |
| 68 | // Claims is the claim store: EnsureLocal replaces the journaled |
| 69 | // receipt there after a repair (final review A3). |
| 70 | Claims release.Store |
| 71 | // Participation is main's participation owner for the observed |
| 72 | // operation (productionClaimLocal.Acknowledged): it opens or reuses |
| 73 | // the box's one hostexec controller for {host, op} and returns its |
| 74 | // ready state. A lost lease stays main's denial (CC9). |
| 75 | Participation func(context.Context, hostclaim.Observed) (execctl.State, error) |
| 76 | Identity func(context.Context) (RelocationIdentity, error) |
| 77 | // Producers is the admitted session's drain (ClaimWakeAssembly.Producers). |
| 78 | Producers func(context.Context, hostclaim.Observed) (StopDrainer, error) |
| 79 | // Collect asks root for full stop evidence (CollectClaimStop over the |
| 80 | // instance-control socket); a RecoveryReleaseDrain is submitted as a |
| 81 | // recovery request. |
| 82 | Collect func(context.Context, StopRequest, StopDrainer) (StopResult, error) |
| 83 | // WritersStopped is release step 1, action and observation: main's |
| 84 | // ClaimAssembly.WritersStopped for the claim's host. |
| 85 | WritersStopped func(context.Context, hostclaim.Observed) (ClaimWritersDrained, WritersStoppedEvidence, error) |
| 86 | // Releases is main's relocation-release obligation (claim Plan 11). |
| 87 | Releases RelocationReleaseDischarge |
| 88 | // FlushDB and FlushFiles are release steps 2 and 3. |
| 89 | FlushDB func(context.Context, hostclaim.Observed) (uint64, error) |
| 90 | FlushFiles func(context.Context, hostclaim.Observed) (int64, error) |
| 91 | // Positions is the durable record of the flushed positions. |
| 92 | Positions ReleasePositions |
| 93 | // BuildP, Install, VerifyLocal, Receipt, Admit: Tasks 8b to 8d. |
| 94 | BuildP func(context.Context, hostclaim.Observed) error |
| 95 | Install func(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, error) |
| 96 | VerifyLocal func(context.Context, hostclaim.Observed, hostclaim.LocalInstall) (bool, error) |
| 97 | Receipt func(context.Context, hostclaim.Observed) (hostclaim.LocalInstall, bool, error) |
| 98 | Admit func(context.Context, hostclaim.Observed, bool) (uint64, bool, error) |
| 99 | // Passed is the probe this box already passed for the admitted session, |
| 100 | // from its attempt journal (relocationPassedProbe; final review A2). |
| 101 | Passed func(context.Context, hostclaim.Observed) (uint64, bool, error) |
| 102 | // PreflightKind is root's isolated-launch capability for kind |
| 103 | // relocating (ClaimRoleAssembly.Preflight then PreflightKind; ruling 48). |
| 104 | PreflightKind func(context.Context, hostclaim.Observed) error |
| 105 | // StopProbe is ruling 49's stop of the passed probe session (Task 9). |
| 106 | StopProbe func(context.Context, hostclaim.Observed) (hostclaim.ProbeStop, error) |
| 107 | // ReleaseHold releases this box's row hold for a completed operation it |
| 108 | // serves, and lifts the activator fence StopProbe installed (Task 9). |
| 109 | ReleaseHold func(ctx context.Context, host, op string) error |
| 110 | } |
| 111 | |
| 112 | var _ hostrelocate.Effects = (*RelocationEffects)(nil) |
| 113 | |
| 114 | // Acknowledge returns the lease of main's participation controller for the |
| 115 | // operation (ruling 42). The claim keeps the first acknowledgement; a |
| 116 | // restarted agent's later controller is main's to open. |
| 117 | func (e *RelocationEffects) Acknowledge(ctx context.Context, obs hostclaim.Observed) (hostclaim.ExecutionAcknowledgement, error) { |
| 118 | if e.Participation == nil || obs.Claim == nil || obs.Claim.Holder == nil { |
| 119 | return hostclaim.ExecutionAcknowledgement{}, ErrClaimStopUnavailable |
| 120 | } |
| 121 | st, err := e.Participation(ctx, obs) |
| 122 | if err != nil { |
| 123 | return hostclaim.ExecutionAcknowledgement{}, err |
| 124 | } |
| 125 | if st.Op != obs.Claim.Op.ID || st.Incarnation != obs.Claim.Holder.Incarnation || st.Host != obs.Host { |
| 126 | return hostclaim.ExecutionAcknowledgement{}, execctl.ErrLost |
| 127 | } |
| 128 | return hostclaim.ExecutionAcknowledgement{Op: obs.Claim.Op.ID, Enrollment: st.Enrollment, BootID: st.BootID, ControllerID: st.ControllerID, LeaseETag: st.LeaseETag, Expires: st.Expires}, nil |
| 129 | } |
| 130 | |
| 131 | // Ready checks main's participation controller is active now. A lost one |
| 132 | // is main's retained denial, never reopened here. |
| 133 | func (e *RelocationEffects) Ready(ctx context.Context, obs hostclaim.Observed) error { |
| 134 | _, err := e.Acknowledge(ctx, obs) |
| 135 | return err |
| 136 | } |
| 137 | |
| 138 | // Release runs §5.8's release in claim Plan 11 item 4's order (ruling 28): |
| 139 | // writers stopped and observed, the database and files flushed and the |
| 140 | // positions recorded at the claim as observed; then the obligation queued; |
| 141 | // then root's full stop with the producers' own drain; then Complete. |
| 142 | func (e *RelocationEffects) Release(ctx context.Context, obs hostclaim.Observed) (hostclaim.Release, error) { |
| 143 | if e.Identity == nil || e.Producers == nil || e.Collect == nil || e.WritersStopped == nil || e.Releases == nil || e.FlushDB == nil || e.FlushFiles == nil || e.Positions == nil || obs.Claim == nil || obs.Claim.Holder == nil { |
| 144 | return hostclaim.Release{}, ErrClaimStopUnavailable |
| 145 | } |
| 146 | id, err := e.Identity(ctx) |
| 147 | if err != nil { |
| 148 | return hostclaim.Release{}, err |
| 149 | } |
| 150 | if id.PID <= 0 || id.ProcessStart == "" { |
| 151 | return hostclaim.Release{}, ErrClaimStopUnavailable |
| 152 | } |
| 153 | if key, digest := obs.Claim.Op.Journal["release_stop_receipt"], obs.Claim.Op.Journal["release_stop_sha256"]; key != "" || digest != "" { |
| 154 | return e.replayDischarged(ctx, obs, id, key, digest) |
| 155 | } |
| 156 | h := obs.Claim.Holder |
| 157 | req := releaseStopRequest(obs, id) |
| 158 | recorded, err := e.flushStopped(ctx, obs, id, req) |
| 159 | if err != nil { |
| 160 | return hostclaim.Release{}, err |
| 161 | } |
| 162 | // Root fences a relocation-release stop only while the claim owes it |
| 163 | // (hostguard.StopAuthorityMatches): queued now, at the flushed claim. |
| 164 | if obs, err = e.Releases.Queue(ctx, obs); err != nil { |
| 165 | return hostclaim.Release{}, err |
| 166 | } |
| 167 | req = releaseStopRequest(obs, id) |
| 168 | req.ExpectedDirectoryRevision = recorded.Request.ExpectedDirectoryRevision |
| 169 | var d StopDrainer = RecoveryReleaseDrain{} |
| 170 | if recorded.owner().local(id) { |
| 171 | inner, err := e.Producers(ctx, obs) |
| 172 | if err != nil { |
| 173 | return hostclaim.Release{}, err |
| 174 | } |
| 175 | d = &releaseDrain{inner: inner} |
| 176 | } else { |
| 177 | req.BootID, req.ControllerID = recorded.Request.BootID, recorded.Request.ControllerID |
| 178 | } |
| 179 | result, err := e.Collect(ctx, req, d) |
| 180 | if err != nil { |
| 181 | return hostclaim.Release{}, err |
| 182 | } |
| 183 | if _, err := e.Releases.Complete(ctx, obs, result.Key, result.Digest); err != nil { |
| 184 | return hostclaim.Release{}, err |
| 185 | } |
| 186 | return hostclaim.Release{ReleasedBy: id.Actor.Box, Enrollment: id.Actor.Enrollment, Generation: h.Generation, Session: h.Session, |
| 187 | FinalTXID: recorded.TXID, ManifestSeq: recorded.Seq, Receipt: result.Key, ReceiptSHA256: result.Digest}, nil |
| 188 | } |
| 189 | |
| 190 | // flushStopped is release steps 1 to 3 and the record (ruling 28), the |
| 191 | // one implementation the release and the probe stop share (ruling 87): |
| 192 | // writers stopped and observed for the claim as observed, then the |
| 193 | // database flushed (FlushDB), then the files (FlushFiles), then the |
| 194 | // positions recorded durably for req. A position already recorded for req |
| 195 | // (a retry after a later step failed, or after root's stop closed the |
| 196 | // session's replicator) is returned with nothing done again. Any failure |
| 197 | // returns its error with nothing recorded, so the stop that follows is |
| 198 | // never asked and the next pass flushes again: a flush is never taken as |
| 199 | // done without its record. A context already done refuses before the |
| 200 | // writers are touched. |
| 201 | func (e *RelocationEffects) flushStopped(ctx context.Context, obs hostclaim.Observed, id RelocationIdentity, req StopRequest) (ReleasePosition, error) { |
| 202 | recorded, flushed, err := e.Positions.Recorded(ctx, req) |
| 203 | if err != nil { |
| 204 | return ReleasePosition{}, err |
| 205 | } |
| 206 | if flushed { |
| 207 | return recorded, nil |
| 208 | } |
| 209 | if err := ctx.Err(); err != nil { |
| 210 | return ReleasePosition{}, err |
| 211 | } |
| 212 | h := obs.Claim.Holder |
| 213 | drained, seen, err := e.WritersStopped(ctx, obs) |
| 214 | if err != nil { |
| 215 | return ReleasePosition{}, err |
| 216 | } |
| 217 | if err := drained.For(obs); err != nil { |
| 218 | return ReleasePosition{}, err |
| 219 | } |
| 220 | if seen.Host != obs.Host || seen.Incarnation != h.Incarnation || seen.InstallationStamp != h.Stamp || seen.WriterSession != h.Session || |
| 221 | seen.Box != id.Actor.Box || seen.Enrollment != id.Actor.Enrollment || seen.CheckedAt.IsZero() { |
| 222 | return ReleasePosition{}, fmt.Errorf("%w: writers-stopped evidence names another session", ErrClaimStopIdentity) |
| 223 | } |
| 224 | txid, err := e.FlushDB(ctx, obs) |
| 225 | if err != nil { |
| 226 | return ReleasePosition{}, err |
| 227 | } |
| 228 | seq, err := e.FlushFiles(ctx, obs) |
| 229 | if err != nil { |
| 230 | return ReleasePosition{}, err |
| 231 | } |
| 232 | recorded = ReleasePosition{Request: req, PID: id.PID, ProcessStart: id.ProcessStart, TXID: txid, Seq: seq} |
| 233 | if err := e.Positions.Record(ctx, recorded); err != nil { |
| 234 | return ReleasePosition{}, err |
| 235 | } |
| 236 | return recorded, nil |
| 237 | } |
| 238 | |
| 239 | // probeStopRequest is the probe stop's request for obs's passed probe |
| 240 | // session (ruling 87): the release's request shape at relocating/probing, |
| 241 | // under root's writer-session stop obligation for that session, so its |
| 242 | // recorded positions are never a release's. |
| 243 | func probeStopRequest(obs hostclaim.Observed, id RelocationIdentity) StopRequest { |
| 244 | r := releaseStopRequest(obs, id) |
| 245 | r.ObligationID = hostguard.WriterStopObligation(obs.Claim.Holder.Session) |
| 246 | return r |
| 247 | } |
| 248 | |
| 249 | // probeStopDeadline is ruling 67's bound on the probe stop, as the worker |
| 250 | // counts it: ReleaseDeadline after the probe passed (Probe.At; for an |
| 251 | // untimed probe, probing_since plus ProbeDeadline). A probe or a |
| 252 | // probing_since that cannot be read is an error, never "no bound". |
| 253 | func probeStopDeadline(c *hostclaim.Claim) (time.Time, error) { |
| 254 | var p hostclaim.Probe |
| 255 | if err := json.Unmarshal([]byte(c.Op.Journal["probe"]), &p); err != nil { |
| 256 | return time.Time{}, fmt.Errorf("%w: the passed probe cannot be read: %v", hostrelocate.ErrRefused, err) |
| 257 | } |
| 258 | passed := p.At |
| 259 | if passed.IsZero() { |
| 260 | since, err := time.Parse(time.RFC3339Nano, c.Op.Journal["probing_since"]) |
| 261 | if err != nil { |
| 262 | return time.Time{}, fmt.Errorf("%w: probing_since cannot be read: %v", hostrelocate.ErrRefused, err) |
| 263 | } |
| 264 | passed = since.Add(hostrelocate.ProbeDeadline) |
| 265 | } |
| 266 | return passed.Add(hostrelocate.ReleaseDeadline), nil |
| 267 | } |
| 268 | |
| 269 | // releaseStopRequest is the release's stop request for obs's session, as |
| 270 | // this box's identity id asks it. |
| 271 | func releaseStopRequest(obs hostclaim.Observed, id RelocationIdentity) StopRequest { |
| 272 | c, h := obs.Claim, obs.Claim.Holder |
| 273 | return StopRequest{Region: id.Actor.Region, Host: c.Host, Incarnation: h.Incarnation, InstallationStamp: h.Stamp, WriterSession: h.Session, |
| 274 | OperationID: c.Op.ID, Phase: c.Op.Phase, ObligationID: RelocationReleaseObligation, Box: id.Actor.Box, Enrollment: id.Actor.Enrollment, |
| 275 | BootID: id.BootID, ControllerID: id.ControllerID, ExpectedClaimETag: obs.ETag, ExpectedDirectoryRevision: id.Actor.Binding, RosterRevision: id.Actor.Revision} |
| 276 | } |
| 277 | |
| 278 | // replayDischarged returns the release whose receipt already discharged |
| 279 | // the obligation: the claim journals key and digest and owes the operation |
| 280 | // nothing, the receipt verifies again (main's Complete in its replay form), |
| 281 | // and this box recorded the positions its flush reached for this stop. |
| 282 | // Without a recorded flush it refuses rather than guess, as a replayed |
| 283 | // receipt always does. |
| 284 | func (e *RelocationEffects) replayDischarged(ctx context.Context, obs hostclaim.Observed, id RelocationIdentity, key, digest string) (hostclaim.Release, error) { |
| 285 | c, h := obs.Claim, obs.Claim.Holder |
| 286 | if err := releaseDischarged(c, key, digest); err != nil { |
| 287 | return hostclaim.Release{}, err |
| 288 | } |
| 289 | if err := e.Releases.Verify(ctx, obs, key, digest); err != nil { |
| 290 | return hostclaim.Release{}, err |
| 291 | } |
| 292 | recorded, flushed, err := e.Positions.Recorded(ctx, releaseStopRequest(obs, id)) |
| 293 | if err != nil { |
| 294 | return hostclaim.Release{}, err |
| 295 | } |
| 296 | if !flushed { |
| 297 | return hostclaim.Release{}, fmt.Errorf("%w: a discharged receipt has no recorded flushed positions for this stop", ErrClaimStopReceipt) |
| 298 | } |
| 299 | return hostclaim.Release{ReleasedBy: id.Actor.Box, Enrollment: id.Actor.Enrollment, Generation: h.Generation, Session: h.Session, |
| 300 | FinalTXID: recorded.TXID, ManifestSeq: recorded.Seq, Receipt: key, ReceiptSHA256: digest}, nil |
| 301 | } |
| 302 | |
| 303 | // RecoveryReleaseDrain refuses: a restarted agent's producers are not the |
| 304 | // stop's, and root must replay the durable receipt or nothing. It is |
| 305 | // main's recoveryStopDrain, under the name the Collect provider matches. |
| 306 | type RecoveryReleaseDrain = recoveryStopDrain |
| 307 | |
| 308 | // releaseDrain is the producers' own drain: steps 1 to 3 ran before the |
| 309 | // obligation was queued (ruling 28), so root's collection closes the |
| 310 | // producers and acknowledges, nothing more. |
| 311 | type releaseDrain struct{ inner StopDrainer } |
| 312 | |
| 313 | func (d *releaseDrain) DrainClaim(ctx context.Context, r StopRequest, fence string) (DrainAcknowledgement, error) { |
| 314 | if d.inner == nil { |
| 315 | return DrainAcknowledgement{}, ErrClaimStopUnavailable |
| 316 | } |
| 317 | return d.inner.DrainClaim(ctx, r, fence) |
| 318 | } |
| 319 | |
| 320 | // Retained is empty: main's participation owner keeps and finishes every |
| 321 | // controller (ruling 42), so the participant's Sweep has nothing to do. |
| 322 | func (e *RelocationEffects) Retained() ([]hostrelocate.Participation, error) { return nil, nil } |
| 323 | |
| 324 | // Finish releases the completed operation's row hold on the serving box |
| 325 | // and lifts the probe's activator fence (ruling 49 step 5). The controller |
| 326 | // is main's to close. |
| 327 | func (e *RelocationEffects) Finish(ctx context.Context, host, op string) error { |
| 328 | if e.ReleaseHold == nil { |
| 329 | return ErrClaimStopUnavailable |
| 330 | } |
| 331 | return e.ReleaseHold(ctx, host, op) |
| 332 | } |
| 333 | |
| 334 | func (e *RelocationEffects) Build(ctx context.Context, obs hostclaim.Observed) error { |
| 335 | if e.BuildP == nil { |
| 336 | return ErrClaimStopUnavailable |
| 337 | } |
| 338 | return e.BuildP(ctx, obs) |
| 339 | } |
| 340 | |
| 341 | func (e *RelocationEffects) InstallLocal(ctx context.Context, obs hostclaim.Observed) (hostclaim.LocalInstall, error) { |
| 342 | if e.Install == nil { |
| 343 | return hostclaim.LocalInstall{}, ErrClaimStopUnavailable |
| 344 | } |
| 345 | return e.Install(ctx, obs) |
| 346 | } |
| 347 | |
| 348 | // EnsureLocal is hostrelocate.Effects' check before P's first admission |
| 349 | // (ruling 52): the journaled receipt's install is verified and, when it is |
| 350 | // damaged while its install stamp is still the active one, made whole by |
| 351 | // the same Install provider under that stamp. A retired stamp is never |
| 352 | // reactivated: Install refuses it, and the probe deadline replaces the |
| 353 | // target. An opened P is never reinstalled. |
| 354 | // |
| 355 | // A repair re-renders the held configuration, so its journal, and the |
| 356 | // digest the receipt carries, differ from the journaled receipt's when |
| 357 | // the configuration changed since (final review A3). The receipt is then |
| 358 | // replaced in the claim, in one CAS on a fresh read of the same |
| 359 | // installation (hostclaim.ReplaceLocalInstall), before any admission, so |
| 360 | // this verification, VerifyReceipt and relocationAfterAdmission all |
| 361 | // compare the one digest the box holds with the one the claim journals. A |
| 362 | // CAS that fails leaves the old receipt, which the next pass finds not |
| 363 | // verifiable and repairs again to the same journal. |
| 364 | func (e *RelocationEffects) EnsureLocal(ctx context.Context, obs hostclaim.Observed) error { |
| 365 | if e.VerifyLocal == nil || e.Install == nil || e.Claims == nil || obs.Claim == nil { |
| 366 | return ErrClaimStopUnavailable |
| 367 | } |
| 368 | c := obs.Claim |
| 369 | if hostclaim.PreparedOpened(c) { |
| 370 | return nil |
| 371 | } |
| 372 | var l hostclaim.LocalInstall |
| 373 | if err := json.Unmarshal([]byte(c.Op.Journal["local_install"]), &l); err != nil || l.InstallStamp == 0 { |
| 374 | return fmt.Errorf("%w: no local install receipt to verify", ErrClaimStopReceipt) |
| 375 | } |
| 376 | if ok, err := e.VerifyLocal(ctx, obs, l); err != nil || ok { |
| 377 | return err |
| 378 | } |
| 379 | got, err := e.Install(ctx, obs) |
| 380 | if err != nil { |
| 381 | return err |
| 382 | } |
| 383 | if got.Box != l.Box || got.Enrollment != l.Enrollment || got.InstallStamp != l.InstallStamp { |
| 384 | return fmt.Errorf("%w: the restored install is not the receipt's", ErrClaimStopReceipt) |
| 385 | } |
| 386 | if got != l { |
| 387 | fresh, err := hostclaim.Read(ctx, e.Claims, obs.Host) |
| 388 | if err != nil { |
| 389 | return err |
| 390 | } |
| 391 | if fresh.Absent || fresh.Claim.Op.ID != c.Op.ID || fresh.Claim.Op.Journal["prepared"] != c.Op.Journal["prepared"] { |
| 392 | return fmt.Errorf("%w: the installation changed under the repair", hostrelocate.ErrRefused) |
| 393 | } |
| 394 | if obs, err = hostclaim.Update(ctx, e.Claims, fresh, hostclaim.ReplaceLocalInstall(c.Op.ID, l, got)); err != nil { |
| 395 | return err |
| 396 | } |
| 397 | l = got |
| 398 | } |
| 399 | ok, err := e.VerifyLocal(ctx, obs, l) |
| 400 | if err == nil && !ok { |
| 401 | err = fmt.Errorf("%w: the restored install did not verify", ErrClaimStopReceipt) |
| 402 | } |
| 403 | return err |
| 404 | } |
| 405 | |
| 406 | // VerifyReceipt is hostrelocate.Effects' check before every admission |
| 407 | // after P is opened: the box's durable receipt must be the one the claim |
| 408 | // journaled, exactly. It reads the receipt alone: the files, env and |
| 409 | // timers are the admitted writers' now, and nothing is restored. |
| 410 | func (e *RelocationEffects) VerifyReceipt(ctx context.Context, obs hostclaim.Observed) (bool, error) { |
| 411 | if e.Receipt == nil || obs.Claim == nil { |
| 412 | return false, ErrClaimStopUnavailable |
| 413 | } |
| 414 | var want hostclaim.LocalInstall |
| 415 | if err := json.Unmarshal([]byte(obs.Claim.Op.Journal["local_install"]), &want); err != nil || want.InstallStamp == 0 { |
| 416 | return false, nil // no journaled receipt to match: nothing vouches for the local state |
| 417 | } |
| 418 | got, ok, err := e.Receipt(ctx, obs) |
| 419 | if err != nil { |
| 420 | return false, err |
| 421 | } |
| 422 | return ok && got == want, nil |
| 423 | } |
| 424 | |
| 425 | func (e *RelocationEffects) Probe(ctx context.Context, obs hostclaim.Observed, certified bool) (uint64, bool, error) { |
| 426 | if e.Admit == nil { |
| 427 | return 0, false, ErrClaimStopUnavailable |
| 428 | } |
| 429 | return e.Admit(ctx, obs, certified) |
| 430 | } |
| 431 | |
| 432 | // PassedProbe recovers a passed probe whose record never reached the claim |
| 433 | // (final review A2) from the box's durable attempt journal, keyed by the |
| 434 | // admitted session. |
| 435 | func (e *RelocationEffects) PassedProbe(ctx context.Context, obs hostclaim.Observed) (uint64, bool, error) { |
| 436 | if e.Passed == nil { |
| 437 | return 0, false, ErrClaimStopUnavailable |
| 438 | } |
| 439 | return e.Passed(ctx, obs) |
| 440 | } |
| 441 | |
| 442 | // Preflight refuses before an attempt is counted while root cannot launch |
| 443 | // a relocating probe (ruling 48). |
| 444 | func (e *RelocationEffects) Preflight(ctx context.Context, obs hostclaim.Observed) error { |
| 445 | if e.PreflightKind == nil || obs.Claim == nil || obs.Claim.Op.Kind != hostclaim.KindRelocating { |
| 446 | return ErrClaimStopUnavailable |
| 447 | } |
| 448 | return e.PreflightKind(ctx, obs) |
| 449 | } |
| 450 | |
| 451 | // EndProbe is ruling 49's stop of the passed probe session, flushed first |
| 452 | // (ruling 87): the session's writers are stopped, its database and files |
| 453 | // flushed and the positions recorded (flushStopped, the release's own |
| 454 | // steps), and only then the probe stop (StopProbe: the activator's owned |
| 455 | // fence and stop, which closes the session's replicator, then root's |
| 456 | // writer-session stop). A write the probe session acknowledged is then |
| 457 | // shipped before the session ends, so the first serving wake's cold |
| 458 | // admission, whose writes_lost is bounded by the remote's MaxTXID, loses |
| 459 | // none of it. The flush is bounded by the stop's own deadline (ruling 67, |
| 460 | // probeStopDeadline): past it the flush refuses and the worker marks the |
| 461 | // op stuck. A failed flush refuses the stop with nothing asked of the |
| 462 | // activator or root, and the next pass flushes again. |
| 463 | func (e *RelocationEffects) EndProbe(ctx context.Context, obs hostclaim.Observed) (hostclaim.ProbeStop, error) { |
| 464 | if e.StopProbe == nil || e.Identity == nil || e.WritersStopped == nil || e.FlushDB == nil || e.FlushFiles == nil || e.Positions == nil || obs.Claim == nil || obs.Claim.Holder == nil { |
| 465 | return hostclaim.ProbeStop{}, ErrClaimStopUnavailable |
| 466 | } |
| 467 | id, err := e.Identity(ctx) |
| 468 | if err != nil { |
| 469 | return hostclaim.ProbeStop{}, err |
| 470 | } |
| 471 | if id.PID <= 0 || id.ProcessStart == "" { |
| 472 | return hostclaim.ProbeStop{}, ErrClaimStopUnavailable |
| 473 | } |
| 474 | // Only this box's passed probe session is flushed: its writers are |
| 475 | // stopped below, so another session is never touched. |
| 476 | if err := passedProbeSession(obs.Claim, id.Actor.Box); err != nil { |
| 477 | return hostclaim.ProbeStop{}, err |
| 478 | } |
| 479 | deadline, err := probeStopDeadline(obs.Claim) |
| 480 | if err != nil { |
| 481 | return hostclaim.ProbeStop{}, err |
| 482 | } |
| 483 | bounded, cancel := context.WithDeadline(ctx, deadline) |
| 484 | _, err = e.flushStopped(bounded, obs, id, probeStopRequest(obs, id)) |
| 485 | cancel() |
| 486 | if err != nil { |
| 487 | return hostclaim.ProbeStop{}, err |
| 488 | } |
| 489 | return e.StopProbe(ctx, obs) |
| 490 | } |
| 491 | |
amadan