| 1 | package hostclaim |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "github.com/carlosframework/platform/internal/regionname" |
| 6 | "reflect" |
| 7 | "strconv" |
| 8 | ) |
| 9 | |
| 10 | const RootGeneration = "@root" |
| 11 | |
| 12 | type Admission struct { |
| 13 | ObservedOpID, ObservedPhase string // Empty values are compared literally. |
| 14 | ObservedSession uint64 |
| 15 | ObservedGeneration string |
| 16 | Box string |
| 17 | Mode string |
| 18 | NewGeneration string |
| 19 | PredecessorCertified bool |
| 20 | RemoteMaxTXID uint64 |
| 21 | KeepStamp bool |
| 22 | Coverage string |
| 23 | Region string |
| 24 | PredecessorRegion string |
| 25 | AdditionalWritesLost []WritesLost |
| 26 | LossID string |
| 27 | lossVerified bool |
| 28 | fenceVerified bool |
| 29 | } |
| 30 | |
| 31 | // AdmitWriter is §6.4a's writer admission CAS within the current operation. |
| 32 | // Warm admission keeps the installation; cold admission installs a fresh |
| 33 | // generation and stamp. The caller checks the evidence required for warm mode |
| 34 | // and supplies PredecessorCertified. |
| 35 | func AdmitWriter(in Admission) func(*Claim) error { |
| 36 | return func(c *Claim) error { |
| 37 | if in.Region == "" { |
| 38 | return fmt.Errorf("%w: an admission names its region", ErrWrongState) |
| 39 | } |
| 40 | if c.Holder == nil || c.Op.Exec == "" || c.Op.Exec != in.Box { |
| 41 | return fmt.Errorf("%w: only the executor may admit a writer", ErrWrongState) |
| 42 | } |
| 43 | if c.Op.ID != in.ObservedOpID || c.Op.Phase != in.ObservedPhase { |
| 44 | return fmt.Errorf("%w: operation or phase moved since it was observed", ErrWrongState) |
| 45 | } |
| 46 | switch c.Op.Kind { |
| 47 | case KindCreating, KindDeleting, KindParking, KindPurging, KindPlacing: |
| 48 | return fmt.Errorf("%w: op kind %q cannot admit a writer", ErrWrongState, c.Op.Kind) |
| 49 | case KindRelocating: |
| 50 | if err := relocatingAdmission(c, in); err != nil { |
| 51 | return err |
| 52 | } |
| 53 | if err := relocationProbeOpening(c, in); err != nil { |
| 54 | return err |
| 55 | } |
| 56 | } |
| 57 | id, err := OperationLossID(c) |
| 58 | if err != nil { |
| 59 | return err |
| 60 | } |
| 61 | if id != "" && (in.Coverage != "" || (!OperationLossConsumed(c, id) && (!in.lossVerified || in.LossID != id))) { |
| 62 | return ErrLossRecord |
| 63 | } |
| 64 | |
| 65 | if c.Op.Kind == KindRestoring && c.Op.Journal["fence."+c.Op.Intent["target"]] != "" && in.Coverage != "" { |
| 66 | return ErrFence |
| 67 | } |
| 68 | if _, pending := PendingMoveFenceLoss(c); pending && !in.fenceVerified { |
| 69 | return ErrFence |
| 70 | } |
| 71 | h := c.Holder |
| 72 | if in.LossID != "" && !in.lossVerified { |
| 73 | return fmt.Errorf("%w: declaration-bound admission requires verified durable declaration", ErrWrongState) |
| 74 | } |
| 75 | if in.LossID != "" { |
| 76 | for _, prior := range h.Admissions { |
| 77 | if prior.LossID != in.LossID { |
| 78 | continue |
| 79 | } |
| 80 | if prior.Box == in.Box && prior.Mode == in.Mode && prior.Generation == in.NewGeneration && |
| 81 | reflect.DeepEqual(prior.AdditionalWritesLost, in.AdditionalWritesLost) { |
| 82 | return ErrLossAlreadyApplied |
| 83 | } |
| 84 | return fmt.Errorf("%w: declaration id already used by another admission", ErrWrongState) |
| 85 | } |
| 86 | } |
| 87 | if c.Op.Kind == KindServing && h.WakeRetry != nil { |
| 88 | r := h.WakeRetry |
| 89 | if r.State != "attempting" || r.Attempt == nil || r.Attempt.Session != 0 || r.Attempt.Box != in.Box { |
| 90 | return fmt.Errorf("%w: serving writer admission has no matching retry reservation", ErrWrongState) |
| 91 | } |
| 92 | } |
| 93 | if h.Session != in.ObservedSession || h.Generation != in.ObservedGeneration { |
| 94 | return fmt.Errorf("%w: writer session or generation moved since it was observed", ErrWrongState) |
| 95 | } |
| 96 | if in.KeepStamp && (in.Mode != "cold" || h.Stamp <= h.Session) { |
| 97 | return fmt.Errorf("%w: keeping a stamp requires cold admission before its first writer", ErrWrongState) |
| 98 | } |
| 99 | switch in.Mode { |
| 100 | case "warm": |
| 101 | if !in.PredecessorCertified || in.NewGeneration != "" { |
| 102 | return fmt.Errorf("%w: a warm admission requires a certificate and keeps the generation", ErrWrongState) |
| 103 | } |
| 104 | case "cold": |
| 105 | if in.NewGeneration == "" || in.NewGeneration == h.Generation { |
| 106 | return fmt.Errorf("%w: a cold recovery needs a new generation", ErrWrongState) |
| 107 | } |
| 108 | default: |
| 109 | return fmt.Errorf("%w: unknown writer admission mode %q", ErrWrongState, in.Mode) |
| 110 | } |
| 111 | |
| 112 | switch in.Coverage { |
| 113 | case "", "sealed", "clean-return": |
| 114 | default: |
| 115 | return ErrWrongState |
| 116 | } |
| 117 | predecessor := AdmissionRecord{Session: h.Session, Box: h.WriterBox, Generation: h.Generation, Region: in.PredecessorRegion} |
| 118 | if len(h.Admissions) > 0 { |
| 119 | predecessor = h.Admissions[len(h.Admissions)-1] |
| 120 | } |
| 121 | if in.Mode == "warm" && !regionname.SameRegion(c.Op.Intent["home_region"], in.Region, predecessor.Region) { |
| 122 | return fmt.Errorf("%w: a warm admission stays in its predecessor's region", ErrWrongState) |
| 123 | } |
| 124 | type writerSession struct { |
| 125 | box string |
| 126 | session uint64 |
| 127 | } |
| 128 | seen := make(map[writerSession]bool, len(in.AdditionalWritesLost)+1) |
| 129 | if in.Mode == "cold" && !in.PredecessorCertified && in.Coverage == "" { |
| 130 | seen[writerSession{predecessor.Box, predecessor.Session}] = true |
| 131 | } |
| 132 | for _, loss := range in.AdditionalWritesLost { |
| 133 | writer := writerSession{loss.Box, loss.Session} |
| 134 | if loss.Box == "" || (loss.Session == 0) != (loss.Generation == "") || seen[writer] { |
| 135 | return ErrWrongState |
| 136 | } |
| 137 | seen[writer] = true |
| 138 | } |
| 139 | entry := AdmissionRecord{ |
| 140 | Session: c.Gen + 1, Box: in.Box, Generation: h.Generation, |
| 141 | LossID: in.LossID, |
| 142 | Predecessor: predecessor.Generation, Mode: in.Mode, Coverage: in.Coverage, |
| 143 | Region: in.Region, PredecessorRegion: predecessor.Region, |
| 144 | Covered: in.Mode == "warm", |
| 145 | AdditionalWritesLost: append([]WritesLost(nil), in.AdditionalWritesLost...), |
| 146 | } |
| 147 | for _, ob := range c.Obligations { |
| 148 | // Only a park of the holder's whole location retires the |
| 149 | // predecessor; a scoped park of another location (an old |
| 150 | // incarnation's residue, a staging prefix) does not. |
| 151 | if ob.Kind == "park-replica" && regionname.SameRegion(c.Op.Intent["home_region"], ob.Region, entry.PredecessorRegion) && |
| 152 | (ob.Park == nil || ob.Park.Location == h.Location) { |
| 153 | entry.Covered = true |
| 154 | break |
| 155 | } |
| 156 | } |
| 157 | if in.Mode == "cold" { |
| 158 | entry.Generation = in.NewGeneration |
| 159 | if !in.KeepStamp { |
| 160 | h.Stamp = entry.Session |
| 161 | } |
| 162 | // A predecessor with no writer (a created incarnation's first |
| 163 | // admission: writer_box "") acknowledged nothing, so there is |
| 164 | // nothing to declare; every declaration names a box (§6.4a). |
| 165 | if !in.PredecessorCertified && in.Coverage == "" && predecessor.Box != "" { |
| 166 | entry.WritesLost = &WritesLost{Box: predecessor.Box, Generation: predecessor.Generation, Session: predecessor.Session, RemoteMaxTXID: in.RemoteMaxTXID} |
| 167 | } |
| 168 | } |
| 169 | if err := pendingOrigin(c, entry); err != nil { |
| 170 | return err |
| 171 | } |
| 172 | h.Admissions = append(h.Admissions, entry) |
| 173 | for len(h.Admissions) > 8 && h.Admissions[0].Covered && admissionLossArchived(h.Admissions[0]) { |
| 174 | h.Admissions = h.Admissions[1:] |
| 175 | } |
| 176 | h.Session, h.Generation, h.WriterBox = entry.Session, entry.Generation, in.Box |
| 177 | if in.lossVerified && id != "" { |
| 178 | setJournal(c, "loss_admitted."+id, strconv.FormatUint(entry.Session, 10)) |
| 179 | } |
| 180 | return nil |
| 181 | } |
| 182 | } |
| 183 | |
| 184 | func PublishGeneration(observedOpID, observedPhase string, observedSession uint64, publishedGeneration, region string, newID func() string) func(*Claim) error { |
| 185 | return func(c *Claim) error { |
| 186 | if c.Holder == nil || c.Op.ID != observedOpID || c.Op.Phase != observedPhase || |
| 187 | c.Holder.Session != observedSession || c.Holder.Generation != publishedGeneration || region == "" { |
| 188 | return ErrWrongState |
| 189 | } |
| 190 | switch c.Op.Kind { |
| 191 | case KindCreating, KindDeleting, KindParking, KindPurging, KindPlacing: |
| 192 | return ErrWrongState |
| 193 | case KindRelocating: |
| 194 | // §6.4a item 5: recognizing the kind grants nothing; only a |
| 195 | // phase where a relocating writer session lives may publish. |
| 196 | if !SessionPhase(KindRelocating, c.Op.Phase) { |
| 197 | return ErrWrongState |
| 198 | } |
| 199 | } |
| 200 | |
| 201 | h := c.Holder |
| 202 | admitted := false |
| 203 | for _, record := range h.Admissions { |
| 204 | if record.Generation == publishedGeneration { |
| 205 | admitted = true |
| 206 | break |
| 207 | } |
| 208 | } |
| 209 | if !admitted { |
| 210 | return ErrWrongState |
| 211 | } |
| 212 | activeGeneration := publishedGeneration |
| 213 | if activeGeneration == "" { |
| 214 | activeGeneration = RootGeneration |
| 215 | } |
| 216 | op := c.Op.ID |
| 217 | if op == "" { |
| 218 | op = fmt.Sprintf("session:%d", h.Session) |
| 219 | } |
| 220 | for i := range h.Admissions { |
| 221 | record := &h.Admissions[i] |
| 222 | if record.Covered { |
| 223 | continue |
| 224 | } |
| 225 | predecessor := record.Predecessor |
| 226 | if predecessor == "" { |
| 227 | predecessor = RootGeneration |
| 228 | } |
| 229 | if predecessor == activeGeneration { |
| 230 | continue |
| 231 | } |
| 232 | if record.PredecessorRegion == "" { |
| 233 | record.Covered = true |
| 234 | continue |
| 235 | } |
| 236 | if !regionname.SameRegion(c.Op.Intent["home_region"], record.PredecessorRegion, region) { |
| 237 | continue |
| 238 | } |
| 239 | queued := false |
| 240 | for _, ob := range c.Obligations { |
| 241 | if ob.Kind == "park-generation" && regionname.SameRegion(c.Op.Intent["home_region"], ob.Region, region) && ob.Generation == predecessor { |
| 242 | queued = true |
| 243 | break |
| 244 | } |
| 245 | } |
| 246 | if !queued { |
| 247 | c.Obligations = append(c.Obligations, Obligation{ |
| 248 | ID: newID(), Op: op, Kind: "park-generation", Region: region, |
| 249 | Generation: predecessor, Retirement: newID(), Owner: "region:" + region, |
| 250 | }) |
| 251 | } |
| 252 | record.Covered = true |
| 253 | } |
| 254 | return nil |
| 255 | } |
| 256 | } |
| 257 | |
| 258 | // CoverRegionPredecessors covers predecessors retired by a region's park of |
| 259 | // the whole location. Callers compose it into the CAS that queues that park |
| 260 | // (a committed move's source cleanup). It queues no chores itself. |
| 261 | func CoverRegionPredecessors(observedOpID, observedPhase string, region string) func(*Claim) error { |
| 262 | return func(c *Claim) error { |
| 263 | if c.Holder == nil || c.Op.ID != observedOpID || c.Op.Phase != observedPhase || region == "" { |
| 264 | return ErrWrongState |
| 265 | } |
| 266 | for i := range c.Holder.Admissions { |
| 267 | record := &c.Holder.Admissions[i] |
| 268 | if !record.Covered && regionname.SameRegion(c.Op.Intent["home_region"], record.PredecessorRegion, region) { |
| 269 | record.Covered = true |
| 270 | } |
| 271 | } |
| 272 | return nil |
| 273 | } |
| 274 | } |
| 275 | |
| 276 | // CoverParkedPredecessors covers only the predecessors retired by a sweep. |
| 277 | // Callers compose it into the CAS that queues the sweep and the CAS that |
| 278 | // acknowledges it. It queues no chores itself. |
| 279 | func CoverParkedPredecessors(observedOpID, observedPhase string, region string, generations []string) func(*Claim) error { |
| 280 | parked := make(map[string]bool, len(generations)) |
| 281 | for _, generation := range generations { |
| 282 | parked[generation] = true |
| 283 | } |
| 284 | return func(c *Claim) error { |
| 285 | if c.Holder == nil || c.Op.ID != observedOpID || c.Op.Phase != observedPhase || region == "" { |
| 286 | return ErrWrongState |
| 287 | } |
| 288 | for i := range c.Holder.Admissions { |
| 289 | record := &c.Holder.Admissions[i] |
| 290 | predecessor := record.Predecessor |
| 291 | if predecessor == "" { |
| 292 | predecessor = RootGeneration |
| 293 | } |
| 294 | if !record.Covered && regionname.SameRegion(c.Op.Intent["home_region"], record.PredecessorRegion, region) && parked[predecessor] { |
| 295 | record.Covered = true |
| 296 | } |
| 297 | } |
| 298 | return nil |
| 299 | } |
| 300 | } |
| 301 | |
| 302 | // relocatingAdmission is §4's relocating phase table for a writer admission, |
| 303 | // as claim Plan 11 accepted it (R1): only op.exec's cold probe, in probing, |
| 304 | // of an operation that is not stuck. Every other phase admits no writer. |
| 305 | // The source's admission in starting or releasing needs its admission-fence |
| 306 | // authority, which no claim code supplies yet, so it stays refused; this |
| 307 | // check only ever adds refusals to AdmitWriter's own. |
| 308 | func relocatingAdmission(c *Claim, in Admission) error { |
| 309 | if c.Op.Journal["stuck"] != "" { |
| 310 | return fmt.Errorf("%w: relocation is stuck", ErrWrongState) |
| 311 | } |
| 312 | if c.Op.Phase != PhaseRelocatingProbing { |
| 313 | return fmt.Errorf("%w: relocating/%s admits no writer", ErrWrongState, c.Op.Phase) |
| 314 | } |
| 315 | if in.Mode != "cold" || in.Box != c.Op.Exec || in.Coverage != "" { |
| 316 | return fmt.Errorf("%w: a relocation probe is op.exec's cold admission", ErrWrongState) |
| 317 | } |
| 318 | return nil |
| 319 | } |
| 320 | |
| 321 | // relocationProbeOpening is region HA's prepared generation P on top of |
| 322 | // relocatingAdmission (§5.8, L9, rev 12), and it only adds refusals: the |
| 323 | // probe admission that opens P installs P itself, and P is opened at most |
| 324 | // once. The opening is journaled with the admission, so it outlives the |
| 325 | // admission list's trim. A certified opening must be backed by this |
| 326 | // operation's journaled release, naming the exact predecessor session, and |
| 327 | // a pin reaching that release's final position: that is the only way a |
| 328 | // relocation probe certifies its predecessor (claim Plan 11 grants no |
| 329 | // relocation coverage of its own, so the release is expressed as the |
| 330 | // generic PredecessorCertified, never as a new coverage value). |
| 331 | func relocationProbeOpening(c *Claim, in Admission) error { |
| 332 | p, opened := c.Op.Journal["prepared"], preparedOpened(c) |
| 333 | if p == "" { |
| 334 | // No prepared generation journaled: main's cold probe as claim Plan |
| 335 | // 11 accepted it (region HA's ToProbing always journals P first). |
| 336 | // With no P there is no release to back a certification, so a |
| 337 | // certified relocating probe is refused. |
| 338 | if in.PredecessorCertified { |
| 339 | return fmt.Errorf("%w: a relocating probe certifies only P's opening", ErrWrongState) |
| 340 | } |
| 341 | return nil |
| 342 | } |
| 343 | if (!opened && in.NewGeneration != p) || (opened && in.NewGeneration == p) { |
| 344 | return fmt.Errorf("%w: the first probe opens P, and only it", ErrWrongState) |
| 345 | } |
| 346 | if opened { |
| 347 | if in.PredecessorCertified { |
| 348 | // The release certifies only the session P's opening succeeds. |
| 349 | return fmt.Errorf("%w: only P's opening certifies its predecessor", ErrWrongState) |
| 350 | } |
| 351 | return nil |
| 352 | } |
| 353 | if in.PredecessorCertified { |
| 354 | var r Release |
| 355 | var pin RecoveryPin |
| 356 | if !readJournal(c.Op, "release", &r) || !readJournal(c.Op, "pin", &pin) || pin.MaxTXID < r.FinalTXID || pin.FilesSeq < r.ManifestSeq { |
| 357 | return fmt.Errorf("%w: a certified opening needs this operation's release and a pin reaching it", ErrWrongState) |
| 358 | } |
| 359 | h := c.Holder |
| 360 | predecessor := AdmissionRecord{Session: h.Session, Box: h.WriterBox, Generation: h.Generation} |
| 361 | if len(h.Admissions) > 0 { |
| 362 | predecessor = h.Admissions[len(h.Admissions)-1] |
| 363 | } |
| 364 | if predecessor.Box != r.ReleasedBy || predecessor.Session != r.Session || predecessor.Generation != r.Generation { |
| 365 | return fmt.Errorf("%w: a certified opening names another predecessor", ErrWrongState) |
| 366 | } |
| 367 | } |
| 368 | ensureJournal(c) |
| 369 | c.Op.Journal["prepared_opened"] = p |
| 370 | return nil |
| 371 | } |
| 372 | |
amadan