| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "errors" |
| 7 | "os" |
| 8 | "strings" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/controlsock" |
| 11 | "github.com/carlosframework/platform/internal/hostbox" |
| 12 | "github.com/carlosframework/platform/internal/hostclaim" |
| 13 | "github.com/carlosframework/platform/internal/hostguard" |
| 14 | "github.com/carlosframework/platform/internal/hostwrite" |
| 15 | "github.com/carlosframework/platform/internal/park" |
| 16 | "github.com/carlosframework/platform/internal/regionname" |
| 17 | "github.com/carlosframework/platform/internal/registry" |
| 18 | ) |
| 19 | |
| 20 | // claimStopLocal is this controller's verified identity when it asks for a stop. |
| 21 | type claimStopLocal struct { |
| 22 | Actor hostbox.Actor |
| 23 | BootID string |
| 24 | ControllerID string |
| 25 | PID int |
| 26 | ProcessStart string |
| 27 | } |
| 28 | |
| 29 | // claimLaunchOwner is the tuple that journaled a launch. PID and ProcessStart |
| 30 | // are zero when no record was found; root then refuses the stop. |
| 31 | type claimLaunchOwner struct { |
| 32 | BootID, ControllerID string |
| 33 | DirectoryRevision string |
| 34 | RosterRevision uint64 |
| 35 | PID int |
| 36 | ProcessStart string |
| 37 | } |
| 38 | |
| 39 | // local reports whether this very process wrote the record. The same IDs in a |
| 40 | // restarted process are not local: its producers ended with the old process. |
| 41 | func (o claimLaunchOwner) local(l claimStopLocal) bool { |
| 42 | if o.BootID != l.BootID || o.ControllerID != l.ControllerID { |
| 43 | return false |
| 44 | } |
| 45 | return (o.PID == 0 && o.ProcessStart == "") || (o.PID == l.PID && o.ProcessStart == l.ProcessStart) |
| 46 | } |
| 47 | |
| 48 | func ownerOf(in hostbox.LaunchInventory) claimLaunchOwner { |
| 49 | return claimLaunchOwner{BootID: in.Target.BootID, ControllerID: in.Target.ControllerID, DirectoryRevision: in.Target.DirectoryRevision, RosterRevision: in.Target.RosterRevision, PID: in.ControllerPID, ProcessStart: in.ProcessStart} |
| 50 | } |
| 51 | |
| 52 | // recoveryStopDrain refuses. A recovered controller's producers ended with it, |
| 53 | // and this controller's producers are never offered as that controller's drain. |
| 54 | type recoveryStopDrain struct{} |
| 55 | |
| 56 | func (recoveryStopDrain) DrainClaim(context.Context, StopRequest, string) (DrainAcknowledgement, error) { |
| 57 | return DrainAcknowledgement{}, ErrClaimStopIdentity |
| 58 | } |
| 59 | |
| 60 | func (a ClaimAssembly) bindClaimWakeStop() { |
| 61 | a.Wake.Stop = func(ctx context.Context, captured hostclaim.Observed) error { |
| 62 | if err := a.stopCapturedSession(ctx, captured); err != nil { |
| 63 | return err |
| 64 | } |
| 65 | pending := a.Wake.get(captured.Host) |
| 66 | if pending != nil { |
| 67 | current, _ := pending.snapshot() |
| 68 | // Root has already stopped this exact session. Its pending |
| 69 | // checkpoint may have advanced since capture; either observation |
| 70 | // order identifies the same stopped lease, never a new session. |
| 71 | reaches := a.stopReaches() |
| 72 | if sameStopWriter(current, captured, reaches) || sameStopWriter(captured, current, reaches) { |
| 73 | return a.Wake.release(ctx, captured.Host, pending) |
| 74 | } |
| 75 | } |
| 76 | return nil |
| 77 | } |
| 78 | a.Wake.Retries.StopAttempt = a.stopRetryAttempt |
| 79 | } |
| 80 | |
| 81 | func (a ClaimAssembly) localStopIdentity(ctx context.Context) (claimStopLocal, error) { |
| 82 | if a.stopLocal != nil { |
| 83 | return a.stopLocal(ctx) |
| 84 | } |
| 85 | if a.Controllers == nil { |
| 86 | return claimStopLocal{}, hostwrite.ErrUnavailable |
| 87 | } |
| 88 | actor, err := a.Identity.Verify(ctx) |
| 89 | if err != nil { |
| 90 | return claimStopLocal{}, err |
| 91 | } |
| 92 | controller, err := a.Controllers.VerifiedController(ctx, os.Getpid()) |
| 93 | if err != nil { |
| 94 | return claimStopLocal{}, err |
| 95 | } |
| 96 | boot, err := os.ReadFile("/proc/sys/kernel/random/boot_id") |
| 97 | if err != nil { |
| 98 | return claimStopLocal{}, err |
| 99 | } |
| 100 | start, err := claimConstructionProcessStart() |
| 101 | if err != nil { |
| 102 | return claimStopLocal{}, err |
| 103 | } |
| 104 | local := claimStopLocal{Actor: actor, BootID: strings.TrimSpace(string(boot)), ControllerID: controller, PID: os.Getpid(), ProcessStart: start} |
| 105 | if local.BootID == "" || local.ControllerID == "" || local.ProcessStart == "" { |
| 106 | return claimStopLocal{}, hostwrite.ErrUnavailable |
| 107 | } |
| 108 | return local, nil |
| 109 | } |
| 110 | |
| 111 | // stopRetryAttempt stops the exact tuple persisted in a serving retry attempt. |
| 112 | // After a controller restart or reboot that tuple is not this controller's. |
| 113 | // It is requested in recovery mode, and root proves its controller dead. |
| 114 | func (a ClaimAssembly) stopRetryAttempt(ctx context.Context, o hostclaim.Observed, attempt hostclaim.WakeRetryAttempt) error { |
| 115 | if o.Claim == nil || o.Claim.Holder == nil || attempt.ID == "" || attempt.BootID == "" || attempt.ControllerID == "" { |
| 116 | return hostbox.ErrChanged |
| 117 | } |
| 118 | if a.Wake == nil || a.Launches == nil { |
| 119 | return hostwrite.ErrUnavailable |
| 120 | } |
| 121 | if attempt.Session == 0 { |
| 122 | // An unbound attempt cannot reach Prepare, so it never launched. Its |
| 123 | // holder session is stopped from that session's own launch records. |
| 124 | if a.Wake.Stop == nil { |
| 125 | return hostwrite.ErrUnavailable |
| 126 | } |
| 127 | return a.Wake.Stop(ctx, o) |
| 128 | } |
| 129 | h := o.Claim.Holder |
| 130 | if h.Session != attempt.Session { |
| 131 | return hostbox.ErrChanged |
| 132 | } |
| 133 | local, err := a.localStopIdentity(ctx) |
| 134 | if err != nil { |
| 135 | return err |
| 136 | } |
| 137 | if attempt.Box != local.Actor.Box || attempt.Enrollment != local.Actor.Enrollment { |
| 138 | return hostbox.ErrChanged |
| 139 | } |
| 140 | owner := claimLaunchOwner{BootID: attempt.BootID, ControllerID: attempt.ControllerID} |
| 141 | recorded, err := a.Launches.Read(ctx, hostguard.FenceTarget{Host: o.Host, Incarnation: h.Incarnation, Stamp: h.Stamp, Session: attempt.Session, Box: attempt.Box, Enrollment: attempt.Enrollment, BootID: attempt.BootID, ControllerID: attempt.ControllerID}) |
| 142 | switch { |
| 143 | case err == nil: |
| 144 | owner = ownerOf(recorded) |
| 145 | // This process already stopped and settled the attempt's own launch |
| 146 | // before failing it; fencing the session would end its later retries. |
| 147 | if owner.local(local) && attemptSettled(recorded, attempt) { |
| 148 | return nil |
| 149 | } |
| 150 | case errors.Is(err, sql.ErrNoRows): |
| 151 | // Keep the persisted tuple. Root refuses it without a launch record; |
| 152 | // this controller's tuple is never substituted. |
| 153 | default: |
| 154 | return err |
| 155 | } |
| 156 | _, err = a.collectCapturedStop(ctx, o, local, owner, false) |
| 157 | return err |
| 158 | } |
| 159 | |
| 160 | func attemptSettled(in hostbox.LaunchInventory, attempt hostclaim.WakeRetryAttempt) bool { |
| 161 | for _, r := range in.Reservations { |
| 162 | if r.ID == attempt.ID+":main" && r.InvocationID == attempt.InvocationID && r.Inputs == attempt.LaunchInputs && r.State == "settled" { |
| 163 | return true |
| 164 | } |
| 165 | } |
| 166 | return false |
| 167 | } |
| 168 | |
| 169 | // stopCapturedSession stops each journaled launch of the captured session. A |
| 170 | // record left by an earlier controller or boot keeps its original tuple. |
| 171 | func (a ClaimAssembly) stopCapturedSession(ctx context.Context, captured hostclaim.Observed) error { |
| 172 | if captured.Claim == nil || captured.Claim.Holder == nil { |
| 173 | return hostbox.ErrChanged |
| 174 | } |
| 175 | h := captured.Claim.Holder |
| 176 | local, err := a.localStopIdentity(ctx) |
| 177 | if err != nil { |
| 178 | return err |
| 179 | } |
| 180 | if a.stopReaches() != nil && claimInitialSourceRecovery(captured, local.Actor.Box) { |
| 181 | if err := (claimCrossBoxHandover{Guard: a.Guard, Evidence: a.Claims, Box: local.Actor.Box}).sourceRecoveryEvidence(ctx, captured); err != nil { |
| 182 | return err |
| 183 | } |
| 184 | } |
| 185 | if h.Session == 0 { |
| 186 | // No writer was ever admitted, so nothing was launched or journaled. |
| 187 | // Root's positive no-process witness (R4) is the stop: this |
| 188 | // controller's tuple, no drain. |
| 189 | _, err = a.collectCapturedStop(ctx, captured, local, claimLaunchOwner{BootID: local.BootID, ControllerID: local.ControllerID}, false) |
| 190 | return err |
| 191 | } |
| 192 | if a.Launches == nil { |
| 193 | return hostwrite.ErrUnavailable |
| 194 | } |
| 195 | launches, err := a.Launches.SessionLaunches(ctx, captured.Host, h.Incarnation, h.Stamp, h.Session, local.Actor.Box, local.Actor.Enrollment) |
| 196 | if err != nil { |
| 197 | return err |
| 198 | } |
| 199 | if len(launches) == 0 { |
| 200 | // Nothing is journaled. Request this controller's tuple; root refuses it |
| 201 | // unless its own records prove the stop. Absence is never emptiness. |
| 202 | _, err = a.collectCapturedStop(ctx, captured, local, claimLaunchOwner{BootID: local.BootID, ControllerID: local.ControllerID}, true) |
| 203 | return err |
| 204 | } |
| 205 | for _, in := range launches { |
| 206 | if _, err := a.collectCapturedStop(ctx, captured, local, ownerOf(in), true); err != nil { |
| 207 | return err |
| 208 | } |
| 209 | } |
| 210 | return nil |
| 211 | } |
| 212 | |
| 213 | // collectRetirementStop stops the one launch tuple represented by a retiring |
| 214 | // route. Retirement can carry only one fence receipt, so multiple journal |
| 215 | // owners are ambiguous and must be refused before any of them is affected. |
| 216 | func (a ClaimAssembly) collectRetirementStop(ctx context.Context, row registry.Route) (hostguard.FenceTarget, error) { |
| 217 | if a.Launches == nil { |
| 218 | return hostguard.FenceTarget{}, hostwrite.ErrUnavailable |
| 219 | } |
| 220 | local, err := a.localStopIdentity(ctx) |
| 221 | if err != nil { |
| 222 | return hostguard.FenceTarget{}, err |
| 223 | } |
| 224 | captured, err := hostclaim.Read(ctx, a.Claims, row.Host) |
| 225 | if err != nil { |
| 226 | return hostguard.FenceTarget{}, err |
| 227 | } |
| 228 | if captured.Claim == nil || captured.Claim.Holder == nil { |
| 229 | return hostguard.FenceTarget{}, hostbox.ErrChanged |
| 230 | } |
| 231 | claim, holder := captured.Claim, captured.Claim.Holder |
| 232 | if holder.Incarnation != row.ClaimIncarnation || holder.Stamp != row.ClaimStamp || claim.Op.Exec != local.Actor.Box { |
| 233 | return hostguard.FenceTarget{}, hostbox.ErrChanged |
| 234 | } |
| 235 | obligation := "" |
| 236 | for _, ob := range claim.Obligations { |
| 237 | if ob.Kind == "park-local" && ob.Box == local.Actor.Box && ob.Owner == "box:"+local.Actor.Box && ob.Op == claim.Op.ID { |
| 238 | if obligation != "" { |
| 239 | return hostguard.FenceTarget{}, hostbox.ErrChanged |
| 240 | } |
| 241 | obligation = ob.ID |
| 242 | } |
| 243 | } |
| 244 | if obligation == "" { |
| 245 | return hostguard.FenceTarget{}, hostbox.ErrChanged |
| 246 | } |
| 247 | launches, err := a.Launches.SessionLaunches(ctx, captured.Host, holder.Incarnation, holder.Stamp, holder.Session, local.Actor.Box, local.Actor.Enrollment) |
| 248 | if err != nil { |
| 249 | return hostguard.FenceTarget{}, err |
| 250 | } |
| 251 | if len(launches) != 1 { |
| 252 | return hostguard.FenceTarget{}, hostbox.ErrChanged |
| 253 | } |
| 254 | result, err := a.collectCapturedStop(ctx, captured, local, ownerOf(launches[0]), false) |
| 255 | if err != nil { |
| 256 | return hostguard.FenceTarget{}, err |
| 257 | } |
| 258 | if verifyRetirementReceipt(ctx, hostguard.FenceReceipt{Target: result.Target, Bytes: result.Bytes, EvidenceAt: result.EvidenceAt}) != nil { |
| 259 | return hostguard.FenceTarget{}, ErrClaimStopReceipt |
| 260 | } |
| 261 | return result.Target, nil |
| 262 | } |
| 263 | |
| 264 | func (a ClaimAssembly) collectCapturedStop(ctx context.Context, captured hostclaim.Observed, local claimStopLocal, owner claimLaunchOwner, writerSession bool) (StopResult, error) { |
| 265 | if captured.Claim == nil || captured.Claim.Holder == nil { |
| 266 | return StopResult{}, hostbox.ErrChanged |
| 267 | } |
| 268 | zero := captured.Claim.Holder.Session == 0 |
| 269 | if zero && !owner.local(local) { |
| 270 | // A session-zero witness is only ever this controller's own request. |
| 271 | return StopResult{}, ErrClaimStopIdentity |
| 272 | } |
| 273 | if owner.BootID == "" || owner.ControllerID == "" { |
| 274 | return StopResult{}, hostwrite.ErrUnavailable |
| 275 | } |
| 276 | observed, err := hostclaim.Read(ctx, a.Claims, captured.Host) |
| 277 | if err != nil { |
| 278 | return StopResult{}, err |
| 279 | } |
| 280 | if !sameStopWriter(captured, observed, a.stopReaches()) { |
| 281 | return StopResult{}, hostbox.ErrChanged |
| 282 | } |
| 283 | actor := local.Actor |
| 284 | claim, holder := observed.Claim, observed.Claim.Holder |
| 285 | home, err := park.VerifiedHome(actor.FleetHomeRegion, actor.HomeRegion, actor.Region, claim.Op.Intent["home_region"]) |
| 286 | if err != nil { |
| 287 | return StopResult{}, hostbox.ErrChanged |
| 288 | } |
| 289 | obligation := "" |
| 290 | if id, ok := a.operationWriterStop(claim, actor); (writerSession || zero) && ok { |
| 291 | obligation = id |
| 292 | } else if claim.Op.Kind != hostclaim.KindServing { |
| 293 | for _, ob := range claim.Obligations { |
| 294 | park := ob.Kind == "park-local" && ob.Owner == "box:"+actor.Box |
| 295 | release := ob.Kind == "relocation-release" && ob.ID == "relocation-release" && ob.Park == nil && regionname.SameRegion(home, ob.Region, actor.Region) && strings.HasPrefix(ob.Owner, "region:") && regionname.SameRegion(home, strings.TrimPrefix(ob.Owner, "region:"), actor.Region) |
| 296 | if (park || release) && ob.Box == actor.Box && ob.Op == claim.Op.ID { |
| 297 | if obligation != "" { |
| 298 | return StopResult{}, hostbox.ErrChanged |
| 299 | } |
| 300 | obligation = ob.ID |
| 301 | } |
| 302 | } |
| 303 | if obligation == "" { |
| 304 | return StopResult{}, hostbox.ErrChanged |
| 305 | } |
| 306 | } |
| 307 | recovering := !owner.local(local) |
| 308 | directoryRevision := owner.DirectoryRevision |
| 309 | if directoryRevision == "" { |
| 310 | // No record was found. Root still requires exact journal evidence. |
| 311 | directoryRevision = actor.Binding |
| 312 | } |
| 313 | // The receipt numeral: the launch record's roster counter, or this |
| 314 | // actor's when no record (or a record from before roster binding) names one. |
| 315 | rosterRevision := owner.RosterRevision |
| 316 | if rosterRevision == 0 { |
| 317 | rosterRevision = actor.Revision |
| 318 | } |
| 319 | req := StopRequest{Region: actor.Region, Recover: recovering, Host: observed.Host, Incarnation: holder.Incarnation, InstallationStamp: holder.Stamp, WriterSession: holder.Session, OperationID: claim.Op.ID, Phase: claim.Op.Phase, ObligationID: obligation, Box: actor.Box, Enrollment: actor.Enrollment, BootID: owner.BootID, ControllerID: owner.ControllerID, ExpectedClaimETag: observed.ETag, ExpectedDirectoryRevision: directoryRevision, RosterRevision: rosterRevision} |
| 320 | target := hostguard.FenceTarget{Region: req.Region, Host: req.Host, Incarnation: req.Incarnation, Stamp: req.InstallationStamp, Session: req.WriterSession, Op: req.OperationID, Phase: req.Phase, ObligationID: req.ObligationID, Box: req.Box} |
| 321 | if zero && claim.Op.Kind == hostclaim.KindServing { |
| 322 | // A never-admitted holder may be a new creator or an adopted legacy |
| 323 | // writer on this box. Root independently proves that no process remains. |
| 324 | if claim.Op.Exec != actor.Box || (holder.WriterBox != "" && holder.WriterBox != actor.Box) || len(holder.Admissions) != 0 { |
| 325 | return StopResult{}, hostbox.ErrChanged |
| 326 | } |
| 327 | } else if !hostguard.StopAuthorityMatches(claim, target) { |
| 328 | return StopResult{}, hostbox.ErrChanged |
| 329 | } |
| 330 | var drain StopDrainer = recoveryStopDrain{} |
| 331 | if !recovering && !zero { |
| 332 | producers, err := a.Wake.Producers(ctx, observed) |
| 333 | if err != nil { |
| 334 | return StopResult{}, err |
| 335 | } |
| 336 | drain = producers |
| 337 | } |
| 338 | var result StopResult |
| 339 | if a.collectStop != nil { |
| 340 | result, err = a.collectStop(ctx, req, drain) |
| 341 | } else { |
| 342 | result, err = CollectClaimStop(ctx, controlsock.Path, req, drain) |
| 343 | } |
| 344 | if err != nil { |
| 345 | return StopResult{}, err |
| 346 | } |
| 347 | return validateTransportStop(req, result) |
| 348 | } |
| 349 | |
| 350 | // sameStopWriter reports whether b, observed at or after a, names a's writer |
| 351 | // session for a stop or a lease release. reaches compares the phases; nil |
| 352 | // requires the same phase, as main did. |
| 353 | func sameStopWriter(a, b hostclaim.Observed, reaches func(kind, from, to string) bool) bool { |
| 354 | return claimSameWriter(a, b, reaches) && a.Claim.Holder.WriterBox == b.Claim.Holder.WriterBox && a.Claim.Holder.Location == b.Claim.Holder.Location |
| 355 | } |
| 356 | |
| 357 | // stopReaches is the teardown's phase predicate once the moves C1 session |
| 358 | // table is bound (Wake.Continue). Stopping a session grants it nothing, so it |
| 359 | // is not M9's adjacency: a probe admitted at probing whose op reached |
| 360 | // committed unseen is still stopped and released as itself (R1). A step back |
| 361 | // never matches. Write authority keeps Wake.Continue. |
| 362 | func (a ClaimAssembly) stopReaches() func(kind, from, to string) bool { |
| 363 | if a.Wake == nil || a.Wake.Continue == nil { |
| 364 | return nil |
| 365 | } |
| 366 | return hostclaim.SessionReaches |
| 367 | } |
| 368 | |
| 369 | // operationWriterStop is the stop authority for this box's admitted operation |
| 370 | // writer, at probing, probed or committed (claim Plan 9 D4): before a |
| 371 | // restarted probe or a cold wake after the open is admitted, the predecessor |
| 372 | // is that writer, and no park-local chore exists or is needed for it. The |
| 373 | // request keeps the op and phase and names the writer session instead of a |
| 374 | // chore; root's full stop and receipt are unchanged, and the cold handover |
| 375 | // then finds that receipt for the predecessor session. Bound only with the |
| 376 | // moves C1 integration (Wake.Continue: CARLOS_HOST_CLAIM and |
| 377 | // CARLOS_MOVES_PHASE1); otherwise main's park-local requirement stands. |
| 378 | func (a ClaimAssembly) operationWriterStop(c *hostclaim.Claim, actor hostbox.Actor) (string, bool) { |
| 379 | if a.stopReaches() == nil { |
| 380 | return "", false |
| 381 | } |
| 382 | if c != nil && c.Op.Kind == hostclaim.KindRelocating { |
| 383 | return hostguard.RelocationWriterStop(c, actor.Box, actor.Region) |
| 384 | } |
| 385 | if c != nil && claimRecoveryKind(c.Op.Kind) { |
| 386 | return hostguard.RecoveryWriterStop(c, actor.Box, actor.Region) |
| 387 | } |
| 388 | return hostguard.LandingWriterStop(c, actor.Box, actor.Region) |
| 389 | } |
| 390 | |
amadan