| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "crypto/sha256" |
| 7 | "encoding/hex" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "log" |
| 12 | "os" |
| 13 | "path/filepath" |
| 14 | "slices" |
| 15 | "sync" |
| 16 | "time" |
| 17 | |
| 18 | "github.com/carlosframework/platform/internal/activator" |
| 19 | "github.com/carlosframework/platform/internal/hostactors" |
| 20 | "github.com/carlosframework/platform/internal/hostbox" |
| 21 | "github.com/carlosframework/platform/internal/hostclaim" |
| 22 | "github.com/carlosframework/platform/internal/hostguard" |
| 23 | "github.com/carlosframework/platform/internal/hostrelocate" |
| 24 | "github.com/carlosframework/platform/internal/hostwrite" |
| 25 | "github.com/carlosframework/platform/internal/lease" |
| 26 | "github.com/carlosframework/platform/internal/moves" |
| 27 | "github.com/carlosframework/platform/internal/park" |
| 28 | "github.com/carlosframework/platform/internal/release" |
| 29 | ) |
| 30 | |
| 31 | // RelocationSource is region HA's ClaimProductionRelocationSource (claim |
| 32 | // Plan 11 R1, R8), registered once at agent startup (Task 10) and opened |
| 33 | // by the production provider only with CARLOS_REGION_HA on. Open returns |
| 34 | // the complete owner set and two refusals (rulings 37, 39), or, while |
| 35 | // moves' C8 provider is not bound (dependency C8, ruling 53), the zero |
| 36 | // owner set: relocating stays refused exactly as before Plan 11. |
| 37 | type RelocationSource struct { |
| 38 | DataDir string |
| 39 | Directory hostactors.Directory |
| 40 | // FilesStore opens the data bucket as a release.Store (ruling 50); |
| 41 | // production is release.NewS3Store. |
| 42 | FilesStore func(ctx context.Context, bucket string) (release.Store, error) |
| 43 | |
| 44 | mu sync.Mutex |
| 45 | serving ClaimServingSources |
| 46 | } |
| 47 | |
| 48 | var _ ClaimProductionRelocationSource = (*RelocationSource)(nil) |
| 49 | |
| 50 | // SetServingSources hands the source the serving launch, markers and the |
| 51 | // C8 provider before Open (claimproduction.go, inside the region HA gate). |
| 52 | func (s *RelocationSource) SetServingSources(src ClaimServingSources) { |
| 53 | s.mu.Lock() |
| 54 | defer s.mu.Unlock() |
| 55 | s.serving = src |
| 56 | } |
| 57 | |
| 58 | func (s *RelocationSource) Open(ctx context.Context, store release.Store, data lease.VersionedStore, actor hostbox.Actor) (ClaimRelocationOwners, []hostbox.PolicyRefusal, error) { |
| 59 | if s == nil { |
| 60 | return ClaimRelocationOwners{}, nil, hostwrite.ErrUnavailable |
| 61 | } |
| 62 | s.mu.Lock() |
| 63 | src := s.serving |
| 64 | s.mu.Unlock() |
| 65 | if store == nil || data == nil || actor.Box == "" || actor.Bucket == "" || actor.Region == "" || s.DataDir == "" || s.FilesStore == nil { |
| 66 | return ClaimRelocationOwners{}, nil, hostwrite.ErrUnavailable |
| 67 | } |
| 68 | if src.Files == nil { |
| 69 | // C8 unbound (ruling 53): no owners, no refusals. The dispatcher |
| 70 | // keeps relocating refused and Task 10's pass stays off. |
| 71 | log.Printf("region HA: relocation owners not opened: the moves C8 files-manifest provider is not bound") |
| 72 | return ClaimRelocationOwners{}, nil, nil |
| 73 | } |
| 74 | if src.Launch == nil || src.Markers == nil || src.Identity == nil { |
| 75 | return ClaimRelocationOwners{}, nil, hostwrite.ErrUnavailable |
| 76 | } |
| 77 | files, err := s.FilesStore(ctx, actor.Bucket) |
| 78 | if err != nil { |
| 79 | return ClaimRelocationOwners{}, nil, err |
| 80 | } |
| 81 | releases := StopReceiptReleases{Receipts: RelocationReleaseReceipts{Store: store, Directory: s.Directory}} |
| 82 | inputs := &RelocationWakeInputs{Launch: src.Launch, FilesManifest: src.Files, Markers: src.Markers, Claims: store, FilesStore: files, |
| 83 | Data: hostbox.RequestStore{Store: data, Check: src.Identity}, Bucket: actor.Bucket, Region: park.PlacementRegion(actor.Region), HomeRegion: actor.HomeRegion, Box: actor.Box, |
| 84 | NewID: newClaimID, Now: time.Now, Releases: releases} |
| 85 | attempts := &relocationProbeAttempts{Claims: store, DataDir: s.DataDir, Box: actor.Box, NewID: newClaimID, Now: time.Now} |
| 86 | owners := ClaimRelocationOwners{ |
| 87 | BeforeAdmission: relocationBeforeAdmission(actor.Box, releases), |
| 88 | AfterAdmission: relocationAfterAdmission(s.DataDir, actor.Box), |
| 89 | ClaimKindOwners: ClaimKindOwners{Inputs: inputs, AdmissionEvidence: inputs.AdmissionEvidence, PublicationEvidence: inputs.PublicationEvidence}, |
| 90 | Operations: attempts, |
| 91 | Policy: relocationPolicy(actor.Box), |
| 92 | } |
| 93 | return owners, []hostbox.PolicyRefusal{refuseStaleStamp, refuseLostBox}, nil |
| 94 | } |
| 95 | |
| 96 | func (s *RelocationSource) Close(context.Context) error { return nil } |
| 97 | |
| 98 | // newClaimID is the 16-byte hex id claimconstruction.go's newID makes. |
| 99 | func newClaimID() string { |
| 100 | var b [16]byte |
| 101 | if _, err := rand.Read(b[:]); err != nil { |
| 102 | return "" |
| 103 | } |
| 104 | return hex.EncodeToString(b[:]) |
| 105 | } |
| 106 | |
| 107 | func refuseStaleStamp(_ context.Context, c *hostclaim.Claim, s hostbox.Selection, _ hostwrite.Identity) error { |
| 108 | if c != nil && c.Op.Kind == hostclaim.KindRelocating && (c.Holder == nil || s.Captured.Incarnation != c.Holder.Incarnation || s.Captured.Stamp != c.Holder.Stamp) { |
| 109 | return hostbox.ErrChanged |
| 110 | } |
| 111 | return nil |
| 112 | } |
| 113 | |
| 114 | func refuseLostBox(_ context.Context, c *hostclaim.Claim, s hostbox.Selection, _ hostwrite.Identity) error { |
| 115 | if c != nil && slices.Contains(c.Lost, s.Box) { |
| 116 | return hostbox.ErrChanged |
| 117 | } |
| 118 | return nil |
| 119 | } |
| 120 | |
| 121 | // relocationBeforeAdmission verifies the source's release (a drain's |
| 122 | // journaled release at its receipt) or its fence (a failover) before the |
| 123 | // target's first admission; a later attempt's predecessor is this box's |
| 124 | // own session, which main's Stop ended. |
| 125 | func relocationBeforeAdmission(box string, releases hostclaim.ReleaseAuthority) func(context.Context, hostclaim.Observed) error { |
| 126 | return func(ctx context.Context, prior hostclaim.Observed) error { |
| 127 | c := prior.Claim |
| 128 | if c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRelocating || c.Op.Exec != box { |
| 129 | return hostbox.ErrChanged |
| 130 | } |
| 131 | if hostclaim.PreparedOpened(c) { |
| 132 | return nil |
| 133 | } |
| 134 | certified, err := hostrelocate.CertifiedFor(c) |
| 135 | if err != nil { |
| 136 | return err // present and unreadable, or mismatched, is never a failover |
| 137 | } |
| 138 | if certified { |
| 139 | var rel hostclaim.Release |
| 140 | if err := json.Unmarshal([]byte(c.Op.Journal["release"]), &rel); err != nil { |
| 141 | return err |
| 142 | } |
| 143 | return releases.VerifyRelease(ctx, *c, rel) |
| 144 | } |
| 145 | src := c.Op.Intent["source_box"] |
| 146 | if slices.Contains(c.Lost, src) && c.Op.Journal["fence."+src] != "" { |
| 147 | return nil |
| 148 | } |
| 149 | return fmt.Errorf("%w: no release and no fence for %s", hostrelocate.ErrRefused, src) |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | // relocationAfterAdmission verifies the target's installation after the |
| 154 | // admission: the local install receipt is this box's, as journaled, and |
| 155 | // P's opening keeps install_stamp. |
| 156 | func relocationAfterAdmission(dataDir, box string) func(context.Context, hostclaim.Observed, hostclaim.Observed) error { |
| 157 | return func(_ context.Context, prior, admitted hostclaim.Observed) error { |
| 158 | c, n := prior.Claim, admitted.Claim |
| 159 | if c == nil || n == nil || n.Holder == nil || n.Op.ID != c.Op.ID || n.Op.Exec != box || n.Holder.WriterBox != box { |
| 160 | return hostbox.ErrChanged |
| 161 | } |
| 162 | var l hostclaim.LocalInstall |
| 163 | if json.Unmarshal([]byte(n.Op.Journal["local_install"]), &l) != nil || l.Box != box || l.InstallStamp == 0 { |
| 164 | return hostbox.ErrChanged |
| 165 | } |
| 166 | path, err := relocationInstallPath(dataDir, n.Op.ID) |
| 167 | if err != nil { |
| 168 | return err |
| 169 | } |
| 170 | raw, err := os.ReadFile(path) |
| 171 | if err != nil { |
| 172 | return err |
| 173 | } |
| 174 | if sum := sha256.Sum256(raw); hex.EncodeToString(sum[:]) != l.JournalSHA256 { |
| 175 | return hostbox.ErrChanged |
| 176 | } |
| 177 | if n.Holder.Generation == n.Op.Journal["prepared"] && n.Holder.Stamp != l.InstallStamp { |
| 178 | return hostbox.ErrChanged |
| 179 | } |
| 180 | return nil |
| 181 | } |
| 182 | } |
| 183 | |
| 184 | // relocationProbe is what one probe attempt over the claim needs and what |
| 185 | // Preflight reads before the attempt is counted (ruling 48; review M1): |
| 186 | // the journaled release's verdict, the prepared authority, the pin's |
| 187 | // agreement with it, and when probing began. Each is a deterministic |
| 188 | // reading of the claim, so a defect never heals by itself: it is found |
| 189 | // before an attempt is spent on it, and found again at the attempt. |
| 190 | type relocationProbe struct { |
| 191 | certified bool |
| 192 | auth hostclaim.PreparedAuthority |
| 193 | empty bool |
| 194 | since time.Time |
| 195 | } |
| 196 | |
| 197 | func relocationProbeClaim(c *hostclaim.Claim) (relocationProbe, error) { |
| 198 | if c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRelocating { |
| 199 | return relocationProbe{}, hostbox.ErrChanged |
| 200 | } |
| 201 | certified, err := hostrelocate.CertifiedFor(c) |
| 202 | if err != nil { |
| 203 | return relocationProbe{}, err |
| 204 | } |
| 205 | auth, err := preparedAuthority(c) |
| 206 | if err != nil { |
| 207 | return relocationProbe{}, err |
| 208 | } |
| 209 | empty, err := emptyPrepared(c, auth) |
| 210 | if err != nil { |
| 211 | return relocationProbe{}, err |
| 212 | } |
| 213 | since, err := time.Parse(time.RFC3339Nano, c.Op.Journal["probing_since"]) |
| 214 | if err != nil { |
| 215 | return relocationProbe{}, fmt.Errorf("%w: probing_since: %v", hostbox.ErrChanged, err) |
| 216 | } |
| 217 | return relocationProbe{certified: certified, auth: auth, empty: empty, since: since}, nil |
| 218 | } |
| 219 | |
| 220 | // relocationAdmit is one probe attempt: the activator's ProbeWake with the |
| 221 | // probe deadline and, on P's prepared opening, P's install manifest. The |
| 222 | // claim it reads was checked by Preflight before the attempt was counted |
| 223 | // (relocationProbeClaim); it is read again here, on the counted |
| 224 | // observation. The install manifest read and a ProbeWake error that is |
| 225 | // not a probe result (an activator or launch failure) run after the count, |
| 226 | // so an attempt spent on transient I/O is an attempt all the same (final |
| 227 | // review minor 11; the same shape as moves' probe): until claim ask C11 |
| 228 | // one attempt is one launch, and its follow-up (plan "After this plan", |
| 229 | // C11) is where three attempts become three probes by moving these reads |
| 230 | // before the count. |
| 231 | func (a ClaimAssembly) relocationAdmit(cfg RelocationAssemblyConfig) func(context.Context, hostclaim.Observed, bool) (uint64, bool, error) { |
| 232 | return func(ctx context.Context, o hostclaim.Observed, certified bool) (uint64, bool, error) { |
| 233 | if cfg.Activator == nil || cfg.Versions == nil || o.Claim == nil || o.Claim.Holder == nil { |
| 234 | return 0, false, ErrClaimStopUnavailable |
| 235 | } |
| 236 | c := o.Claim |
| 237 | probe, err := relocationProbeClaim(c) |
| 238 | if err != nil { |
| 239 | return 0, false, err |
| 240 | } |
| 241 | if certified != probe.certified { |
| 242 | return 0, false, fmt.Errorf("%w: the participant and the owners disagree on certification", hostrelocate.ErrRefused) |
| 243 | } |
| 244 | pc := activator.ProbeConfig{Host: o.Host, Op: c.Op.ID, HoldOp: c.Op.ID, Deadline: probe.since.Add(hostrelocate.ProbeDeadline)} |
| 245 | if !hostclaim.PreparedOpened(c) && !probe.empty { |
| 246 | im, _, err := readInstallManifest(ctx, cfg.Versions, cfg.Bucket, moves.GenerationPrefix(c.Holder.Location, probe.auth.Generation)) |
| 247 | if err != nil { |
| 248 | return 0, false, err |
| 249 | } |
| 250 | pc.Prepared = &im |
| 251 | } |
| 252 | res, err := cfg.Activator.ProbeWake(ctx, pc) |
| 253 | switch { |
| 254 | case err == nil: |
| 255 | return res.Session, true, nil |
| 256 | case errors.Is(err, activator.ErrProbeFailed), errors.Is(err, activator.ErrAttemptOver): |
| 257 | return 0, false, nil |
| 258 | default: |
| 259 | return 0, false, err |
| 260 | } |
| 261 | } |
| 262 | } |
| 263 | |
| 264 | // relocationStopProbe is ruling 49 steps 1 to 3, while the op is still |
| 265 | // relocating/probing so the captured session still matches: the |
| 266 | // activator's local fence at the session and its stop of the probe entry; |
| 267 | // main's writer-session stop through root (deterministic, so a repeat |
| 268 | // meets the same fence and receipt), which also releases the attempt's |
| 269 | // lease; then root's receipt for exactly that session. |
| 270 | // |
| 271 | // The fence is an owned one (ruling 75, Astra round 6 findings 1 and 2). |
| 272 | // Before anything is asked of the activator, the owner this probe stop |
| 273 | // will install under is persisted with the session in the op's journal |
| 274 | // directory (probe-fence.json); a failed write installs nothing, and the |
| 275 | // next pass writes it again. FenceAndStopAs then installs only on an |
| 276 | // unfenced host, or meets again exactly the fence it installed under |
| 277 | // that owner at that session, deciding under the activator's own lock. |
| 278 | // Any other fence (another operation's, of any generation, including one |
| 279 | // installed after an agent restart emptied the activator's fences while |
| 280 | // the marker survived) is ErrFenceHeld, with nothing stopped and nothing |
| 281 | // asked of root, and is never replaced, so ReleaseHold's lift can never |
| 282 | // reopen admission another operation closed. |
| 283 | func (a ClaimAssembly) relocationStopProbe(cfg RelocationAssemblyConfig) func(context.Context, hostclaim.Observed) (hostclaim.ProbeStop, error) { |
| 284 | return func(ctx context.Context, o hostclaim.Observed) (hostclaim.ProbeStop, error) { |
| 285 | if cfg.Activator == nil || a.Wake == nil || a.Wake.Stop == nil || o.Claim == nil || o.Claim.Holder == nil { |
| 286 | return hostclaim.ProbeStop{}, ErrClaimStopUnavailable |
| 287 | } |
| 288 | local, err := a.localStopIdentity(ctx) |
| 289 | if err != nil { |
| 290 | return hostclaim.ProbeStop{}, err |
| 291 | } |
| 292 | c, h, box := o.Claim, o.Claim.Holder, local.Actor.Box |
| 293 | if err := passedProbeSession(c, box); err != nil { |
| 294 | return hostclaim.ProbeStop{}, err |
| 295 | } |
| 296 | marker, err := relocationProbeFencePath(cfg.DataDir, c.Op.ID) |
| 297 | if err != nil { |
| 298 | return hostclaim.ProbeStop{}, err |
| 299 | } |
| 300 | owner, err := relocationProbeFenceOwner(marker, o.Host, h.Session) |
| 301 | if err != nil { |
| 302 | return hostclaim.ProbeStop{}, err // nothing installed: the next pass persists it again |
| 303 | } |
| 304 | fenced, cancel := context.WithTimeout(ctx, hostbox.RequestLimit) |
| 305 | _, err = cfg.Activator.FenceAndStopAs(fenced, o.Host, h.Session, owner) |
| 306 | cancel() |
| 307 | if err != nil && !errors.Is(err, activator.ErrStopUnproven) { |
| 308 | return hostclaim.ProbeStop{}, err // not this probe's fence: nothing stopped, nothing asked |
| 309 | } |
| 310 | if err := a.Wake.Stop(ctx, o); err != nil { |
| 311 | return hostclaim.ProbeStop{}, err |
| 312 | } |
| 313 | r, err := writerSessionReceipt(ctx, a.Guard, o.Host, hostguard.Captured{Incarnation: h.Incarnation, Stamp: h.Stamp}, c.Op.ID, box, h.Session) |
| 314 | if err != nil { |
| 315 | return hostclaim.ProbeStop{}, err |
| 316 | } |
| 317 | sum := sha256.Sum256(r.Bytes) |
| 318 | return hostclaim.ProbeStop{Box: box, Session: h.Session, Receipt: r.Target.FenceID, ReceiptSHA256: hex.EncodeToString(sum[:])}, nil |
| 319 | } |
| 320 | } |
| 321 | |
| 322 | // passedProbeSession checks c is this box's passed probe at |
| 323 | // relocating/probing, its session the holder's own: the one session the |
| 324 | // probe stop may flush (ruling 87) and stop (ruling 49). Anything else is |
| 325 | // hostbox.ErrChanged, with nothing stopped. |
| 326 | func passedProbeSession(c *hostclaim.Claim, box string) error { |
| 327 | var p hostclaim.Probe |
| 328 | if c == nil || c.Holder == nil || box == "" || json.Unmarshal([]byte(c.Op.Journal["probe"]), &p) != nil || p.Box != box || p.Session != c.Holder.Session || |
| 329 | c.Op.Exec != box || c.Op.Phase != hostclaim.PhaseRelocatingProbing || c.Holder.WriterBox != box { |
| 330 | return hostbox.ErrChanged |
| 331 | } |
| 332 | return nil |
| 333 | } |
| 334 | |
| 335 | // relocationProbeFencePath is where this box's probe stop persists the |
| 336 | // owner of the activator fence it installs, with the session and host: |
| 337 | // <DataDir>/relocate/ops/<op>/probe-fence.json. The journal sweep keeps |
| 338 | // the op's directory until the completed relocation is marked discharged |
| 339 | // (ruling 76), so the lift always finds it. |
| 340 | func relocationProbeFencePath(dataDir, op string) (string, error) { |
| 341 | dir, err := relocationOpDir(dataDir, op) |
| 342 | if err != nil { |
| 343 | return "", err |
| 344 | } |
| 345 | return filepath.Join(dir, "probe-fence.json"), nil |
| 346 | } |
| 347 | |
| 348 | // relocationProbeFence is the persisted owner of one probe stop's fence. |
| 349 | type relocationProbeFence struct { |
| 350 | Host string `json:"host"` |
| 351 | Session uint64 `json:"session"` |
| 352 | Owner string `json:"owner"` |
| 353 | } |
| 354 | |
| 355 | // relocationProbeFenceOwner returns the owner persisted for host's probe |
| 356 | // session, or persists a new, random one first. A marker that is present |
| 357 | // is the only record of the owner a fence may already be installed under, |
| 358 | // so it is never replaced: one that cannot be read, or that names another |
| 359 | // host or session, refuses (absence is never evidence), nothing is |
| 360 | // installed, and the marker stays for an operator. The op's probe session |
| 361 | // is fixed once its probe passed, so such a marker is a defect, never a |
| 362 | // retry's. |
| 363 | func relocationProbeFenceOwner(path, host string, session uint64) (string, error) { |
| 364 | rec, found, err := readRelocationProbeFence(path) |
| 365 | if err != nil { |
| 366 | return "", err |
| 367 | } |
| 368 | if found { |
| 369 | if rec.Host != host || rec.Session != session { |
| 370 | return "", fmt.Errorf("%w: %s names %s session %d, not %s session %d", hostrelocate.ErrRefused, path, rec.Host, rec.Session, host, session) |
| 371 | } |
| 372 | return rec.Owner, nil |
| 373 | } |
| 374 | var b [16]byte |
| 375 | if _, err := rand.Read(b[:]); err != nil { |
| 376 | return "", err |
| 377 | } |
| 378 | rec = relocationProbeFence{Host: host, Session: session, Owner: hex.EncodeToString(b[:])} |
| 379 | raw, err := json.Marshal(rec) |
| 380 | if err != nil { |
| 381 | return "", err |
| 382 | } |
| 383 | if err := writeRelocationJournal(path, raw); err != nil { |
| 384 | return "", err |
| 385 | } |
| 386 | return rec.Owner, nil |
| 387 | } |
| 388 | |
| 389 | // readRelocationProbeFence reads the persisted owner; false, with no |
| 390 | // error, only when there is no marker at all. A marker that does not |
| 391 | // decode, or names no host, session or owner, is an error, never "none". |
| 392 | func readRelocationProbeFence(path string) (relocationProbeFence, bool, error) { |
| 393 | raw, err := os.ReadFile(path) |
| 394 | if errors.Is(err, os.ErrNotExist) { |
| 395 | return relocationProbeFence{}, false, nil |
| 396 | } |
| 397 | if err != nil { |
| 398 | return relocationProbeFence{}, false, err |
| 399 | } |
| 400 | var rec relocationProbeFence |
| 401 | if err := json.Unmarshal(raw, &rec); err != nil { |
| 402 | return relocationProbeFence{}, false, fmt.Errorf("%w: %s cannot be read: %v", hostrelocate.ErrRefused, path, err) |
| 403 | } |
| 404 | if rec.Host == "" || rec.Session == 0 || rec.Owner == "" { |
| 405 | return relocationProbeFence{}, false, fmt.Errorf("%w: %s names no host, session or owner", hostrelocate.ErrRefused, path) |
| 406 | } |
| 407 | return rec, true, nil |
| 408 | } |
| 409 | |
| 410 | // relocationReleaseHold finishes a completed relocation on the box it |
| 411 | // completed on (ruling 49 step 5, rulings 76 and 80), in this order: while |
| 412 | // the row is held for the op, the holder must still be the stopped probe |
| 413 | // session, main's participation is discharged (C10) and the hold is |
| 414 | // released (ruling 66: nothing is released before the discharge); then |
| 415 | // the activator fence the probe stop installed under its persisted owner |
| 416 | // is lifted; last, the serving op's completed relocation is marked |
| 417 | // discharged, which is what lets any successor start (the completion |
| 418 | // barrier, ruling 77). The pass calls it for every serving claim whose |
| 419 | // completion record names this box and is not yet marked, never from the |
| 420 | // row (ruling 80), and every step is idempotent, so a failure or a crash |
| 421 | // at any point leaves the record unmarked and the next pass resumes where |
| 422 | // it stopped. Once the row's hold is released, a replay discharges again |
| 423 | // rather than inferring that the discharge ran (absence is never |
| 424 | // evidence): main's discharge then finds no slot and no .hostexec record, |
| 425 | // which is positive evidence, and does nothing more; only then are the |
| 426 | // fence lifted and the record marked. A row held by any other op is |
| 427 | // refused with nothing done: nothing has proved this op's participation |
| 428 | // discharged. A missing row is refused: absence is never evidence of the |
| 429 | // discharge. So is a missing, unreadable or mismatched probe-fence owner, |
| 430 | // read before anything is released: the probe stop persisted it before it |
| 431 | // fenced, and without it nothing can say the fence is lifted, so the mark |
| 432 | // is never written over its absence. A claim that is absent, or does not |
| 433 | // serve on this box, is ErrChanged: the pass asks only about completions |
| 434 | // recorded on this box (ruling 80), so a caller that asks about anything |
| 435 | // else is mis-wired, never finished. Anything else leaves everything: an |
| 436 | // excluded box's row is its park-local's. |
| 437 | func (a ClaimAssembly) relocationReleaseHold(cfg RelocationAssemblyConfig) func(context.Context, string, string) error { |
| 438 | return func(ctx context.Context, host, op string) error { |
| 439 | if cfg.Registry == nil || cfg.Activator == nil || cfg.Discharge == nil { |
| 440 | return ErrClaimStopUnavailable |
| 441 | } |
| 442 | local, err := a.localStopIdentity(ctx) |
| 443 | if err != nil { |
| 444 | return err |
| 445 | } |
| 446 | obs, err := hostclaim.Read(ctx, a.Claims, host) |
| 447 | if err != nil { |
| 448 | return err |
| 449 | } |
| 450 | if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil || obs.Claim.Op.Kind != hostclaim.KindServing || obs.Claim.Op.Exec != local.Actor.Box || obs.Claim.Holder.WriterBox != local.Actor.Box { |
| 451 | return fmt.Errorf("%w: %s does not serve on this box", hostbox.ErrChanged, host) |
| 452 | } |
| 453 | done, ok, err := hostclaim.ReadCompletedRelocation(obs.Claim) |
| 454 | if err != nil { |
| 455 | return err // unreadable is not "no completion": nothing is released over it |
| 456 | } |
| 457 | if !ok || done.Op != op || done.Stop.Box != local.Actor.Box { |
| 458 | return fmt.Errorf("%w: %s's serving op does not record %s as completed on this box", hostrelocate.ErrRefused, host, op) |
| 459 | } |
| 460 | if done.Discharged { |
| 461 | return nil |
| 462 | } |
| 463 | row, err := cfg.Registry.Get(host) |
| 464 | if err != nil { |
| 465 | return err |
| 466 | } |
| 467 | if row == nil { |
| 468 | return fmt.Errorf("%w: %s has no row on this box", hostrelocate.ErrRefused, host) |
| 469 | } |
| 470 | if row.HoldOp != "" && row.HoldOp != op { |
| 471 | return fmt.Errorf("%w: %s's row is held by %s, not %s", hostrelocate.ErrRefused, host, row.HoldOp, op) |
| 472 | } |
| 473 | marker, err := relocationProbeFencePath(cfg.DataDir, op) |
| 474 | if err != nil { |
| 475 | return err |
| 476 | } |
| 477 | fence, found, err := readRelocationProbeFence(marker) |
| 478 | if err != nil { |
| 479 | return err // the owner is unreadable: nothing is released, the fence stays, the claim is not marked |
| 480 | } |
| 481 | if !found || fence.Host != host || fence.Session != done.Stop.Session { |
| 482 | return fmt.Errorf("%w: %s has no probe-fence owner for %s's stopped session %d", hostrelocate.ErrRefused, host, op, done.Stop.Session) |
| 483 | } |
| 484 | if row.HoldOp == op { |
| 485 | if done.Stop.Session != obs.Claim.Holder.Session { |
| 486 | return fmt.Errorf("%w: %s's serving holder is not %s's stopped probe session", hostrelocate.ErrRefused, host, op) |
| 487 | } |
| 488 | if err := cfg.Discharge(ctx, obs); err != nil { |
| 489 | return err |
| 490 | } |
| 491 | if _, err := cfg.Registry.ReleaseHold(host, op); err != nil { |
| 492 | return err |
| 493 | } |
| 494 | } else if err := cfg.Discharge(ctx, obs); err != nil { |
| 495 | // The hold is already released. The holder may have moved on |
| 496 | // since, so there is no holder check here: main's serving Check |
| 497 | // let no wake through while a slot or record remained, and the |
| 498 | // discharge's postcondition (no slot, no record) is the |
| 499 | // evidence, not the released hold. |
| 500 | return err |
| 501 | } |
| 502 | cfg.Activator.LiftFenceAs(host, fence.Session, fence.Owner) |
| 503 | _, err = hostclaim.Update(ctx, a.Claims, obs, hostclaim.MarkCompletedDischarged(op)) |
| 504 | if errors.Is(err, hostclaim.ErrAlreadyRecorded) { |
| 505 | return nil |
| 506 | } |
| 507 | return err |
| 508 | } |
| 509 | } |
| 510 | |
amadan