| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "fmt" |
| 7 | "reflect" |
| 8 | "strconv" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/genselect" |
| 11 | "github.com/carlosframework/platform/internal/hostbox" |
| 12 | "github.com/carlosframework/platform/internal/hostclaim" |
| 13 | "github.com/carlosframework/platform/internal/hostops" |
| 14 | "github.com/carlosframework/platform/internal/park" |
| 15 | ) |
| 16 | |
| 17 | // The loss and fence rules of the source's recovery (moves-after-Plan-12 |
| 18 | // Task 4: owed (a) and (b), Plan 4 Task 10's box half; rulings Q6, Q7's |
| 19 | // loss half and Q5 as platform-1e answered it on #236). |
| 20 | // |
| 21 | // A declared or fenced return (Q6). The claim journals a declaration of the |
| 22 | // target's loss under the op (hostclaim.OperationLossID) or a termination |
| 23 | // fence of the target (fence.<target>); the record's rollback cites it |
| 24 | // (Rollback.LossRef: the declaration's record key, or the fence record's |
| 25 | // key). Such a return is clean for want of a target and covered by nothing: |
| 26 | // its probe asks no coverage, for its first admission and every later one, |
| 27 | // since the declaration and the fence stay the op's after the first |
| 28 | // admission consumed them. A record and a claim that disagree, by presence |
| 29 | // or by identity, wait. |
| 30 | // |
| 31 | // The loss baseline validator (Q5, owed (b)). The claim's writer admits the |
| 32 | // first recovery admission under a pending loss through admitLoss or |
| 33 | // admitFencedLoss (hostbox/writer_loss.go), which call LossSources with the |
| 34 | // frozen record. No cross-region read exists (the box role reads only its |
| 35 | // own data bucket; platform-1e): each frozen source is vouched for from the |
| 36 | // maxima already recorded. |
| 37 | // |
| 38 | // - replica:<region>/<gen>, a target session since the cut: the loss |
| 39 | // boundary is the highest TXID of the tuple's own generation that the |
| 40 | // restore contains (platform-1e's loss-boundary rule), never a number |
| 41 | // carried across generations: a fresh target generation restarts its |
| 42 | // TXIDs at [1, 1] (spec §5.1). It is genselect.GenerationMax over the |
| 43 | // restore plan: a generation the plan does not contain (every target |
| 44 | // generation behind a J4 return, which restores src_gen) is 0, every |
| 45 | // write in it lost. Every tuple is kept, however many share 0. The |
| 46 | // claim compares the declared maximum with this validated one |
| 47 | // (hostclaim/lost.go AdmitWriterWithLoss). |
| 48 | // - seal:<move>, the source's cut session when the seal did not cover the |
| 49 | // source's replica: the cut seal the claim journals, its maximum N. |
| 50 | // |
| 51 | // The restore point is checked too: the source baseline's maximum, read |
| 52 | // from its own selected files by genselect.GenerationMax, the rule the |
| 53 | // claim's writer reads. A restore point that is not the recorded source |
| 54 | // baseline, or whose files do not reach its maximum, is a wait |
| 55 | // (hostbox.ErrChanged), within the probe's nine attempts; a frozen source |
| 56 | // this recovery cannot be the owner of is a refusal (hostclaim.ErrLossRecord). |
| 57 | // |
| 58 | // AdmissionEvidence's loss contract (Q7). The writer substitutes the |
| 59 | // validated predecessor's maximum for RemoteMaxTXID and the other validated |
| 60 | // sources for AdditionalWritesLost before it calls AdmissionEvidence. The |
| 61 | // recovery owners accept exactly that substitution, from the same |
| 62 | // LossSources call for the same observation, re-validated; every other |
| 63 | // check of the recovery's evidence, the lineage link and the source |
| 64 | // baseline's record before the CAS included, is unchanged. |
| 65 | |
| 66 | // claimLossRefs are the keys of the target's loss the claim journals under |
| 67 | // the op: the declaration's record key, and the fence record's key. |
| 68 | func claimLossRefs(c *hostclaim.Claim) (map[string]bool, error) { |
| 69 | refs := map[string]bool{} |
| 70 | id, err := hostclaim.OperationLossID(c) |
| 71 | if err != nil { |
| 72 | return nil, err |
| 73 | } |
| 74 | if id != "" { |
| 75 | key, err := hostclaim.LossRecordKey(id) |
| 76 | if err != nil { |
| 77 | return nil, err |
| 78 | } |
| 79 | refs[key] = true |
| 80 | } |
| 81 | if raw := c.Op.Journal["fence."+c.Op.Intent["target"]]; raw != "" { |
| 82 | var ref hostclaim.FenceRef |
| 83 | if err := json.Unmarshal([]byte(raw), &ref); err != nil || ref.Key == "" { |
| 84 | return nil, fmt.Errorf("%w: the target's fence does not read", hostclaim.ErrFence) |
| 85 | } |
| 86 | refs[ref.Key] = true |
| 87 | } |
| 88 | return refs, nil |
| 89 | } |
| 90 | |
| 91 | // recoveryDeclared reports whether a return is a declared or fenced one |
| 92 | // (Q6): the claim journals the target's loss and the record's rollback |
| 93 | // cites it (lossRef). A disagreement, by presence or by identity, is |
| 94 | // hostbox.ErrChanged: the probe waits for the two to agree. |
| 95 | func recoveryDeclared(c *hostclaim.Claim, lossRef string) (bool, error) { |
| 96 | refs, err := claimLossRefs(c) |
| 97 | if err != nil { |
| 98 | return false, fmt.Errorf("%w: the claim's loss of the target: %v", hostbox.ErrChanged, err) |
| 99 | } |
| 100 | if len(refs) == 0 && lossRef == "" { |
| 101 | return false, nil |
| 102 | } |
| 103 | if !refs[lossRef] { |
| 104 | return false, fmt.Errorf("%w: the record and the claim disagree on the target's loss", hostbox.ErrChanged) |
| 105 | } |
| 106 | return true, nil |
| 107 | } |
| 108 | |
| 109 | // lossValidation is one LossSources call's answer, kept for the |
| 110 | // AdmissionEvidence call of the same admission. |
| 111 | type lossValidation struct { |
| 112 | etag string |
| 113 | record hostclaim.LossRecord |
| 114 | valid []hostclaim.ValidatedWriteSource |
| 115 | } |
| 116 | |
| 117 | // LossSources is the moves loss baseline validator |
| 118 | // (hostbox.LossAdmissionSources): the recovery owners' movesLossSources |
| 119 | // for a recovery, refused for every other kind. |
| 120 | func (m *movesProbeOwners) LossSources(ctx context.Context, o hostclaim.Observed, loss hostclaim.LossRecord, in hostclaim.Admission, baseline *genselect.Plan) ([]hostclaim.ValidatedWriteSource, error) { |
| 121 | if !recoveryKind(o) { |
| 122 | return nil, hostbox.ErrUnsupported |
| 123 | } |
| 124 | r := m.recovery() |
| 125 | valid, err := r.movesLossSources(ctx, o, loss, baseline) |
| 126 | if err != nil { |
| 127 | return nil, err |
| 128 | } |
| 129 | m.mu.Lock() |
| 130 | defer m.mu.Unlock() |
| 131 | if m.lossValidated == nil { |
| 132 | m.lossValidated = map[string]lossValidation{} |
| 133 | } |
| 134 | // One per host: a later call for the host replaces an earlier one. |
| 135 | m.lossValidated[o.Host] = lossValidation{etag: o.ETag, record: loss, valid: append([]hostclaim.ValidatedWriteSource(nil), valid...)} |
| 136 | return valid, nil |
| 137 | } |
| 138 | |
| 139 | // movesLossSources vouches for each of loss's frozen sources (see the file |
| 140 | // comment), for the recovery's first admission while loss is the pending |
| 141 | // declaration or the pending fence of this op's target. |
| 142 | func (r movesRecoveryOwners) movesLossSources(ctx context.Context, o hostclaim.Observed, loss hostclaim.LossRecord, baseline *genselect.Plan) ([]hostclaim.ValidatedWriteSource, error) { |
| 143 | v, err := r.recovery(ctx, o) |
| 144 | if err != nil { |
| 145 | return nil, err |
| 146 | } |
| 147 | c := o.Claim |
| 148 | if v.kind != hostclaim.KindRestoring || v.dirty || !v.uncovered || !v.first { |
| 149 | return nil, fmt.Errorf("%w: only a declared or fenced return's first admission declares a loss", hostclaim.ErrLossRecord) |
| 150 | } |
| 151 | if err := r.pendingLoss(c, v, loss); err != nil { |
| 152 | return nil, err |
| 153 | } |
| 154 | if err := r.frozenMaximum(v, baseline); err != nil { |
| 155 | return nil, err |
| 156 | } |
| 157 | cut, err := strconv.ParseUint(v.cut, 10, 64) |
| 158 | if err != nil { |
| 159 | return nil, fmt.Errorf("%w: the cut's session is not journalled", hostbox.ErrChanged) |
| 160 | } |
| 161 | out := make([]hostclaim.ValidatedWriteSource, 0, len(loss.WriteSources)) |
| 162 | for _, s := range loss.WriteSources { |
| 163 | max, err := r.sourceMaximum(c, v, cut, baseline, s) |
| 164 | if err != nil { |
| 165 | return nil, err |
| 166 | } |
| 167 | out = append(out, hostclaim.ValidatedWriteSource{Reason: s.Reason, Box: s.Box, Generation: s.Generation, Session: s.Session, Region: s.Region, Baseline: s.Baseline, RemoteMaxTXID: max}) |
| 168 | } |
| 169 | return out, nil |
| 170 | } |
| 171 | |
| 172 | // pendingLoss requires loss to be what the claim's writer admits next: the |
| 173 | // fence's derived record while the consumed fence's target writer awaits |
| 174 | // its first admission, or else the op's unconsumed declaration; and either |
| 175 | // to be this op's, this holder's, of this op's target, and the one the |
| 176 | // record's rollback cites. |
| 177 | func (r movesRecoveryOwners) pendingLoss(c *hostclaim.Claim, v recoveryProbe, loss hostclaim.LossRecord) error { |
| 178 | refused := func(why string) error { return fmt.Errorf("%w: %s", hostclaim.ErrLossRecord, why) } |
| 179 | if loss.OpID != v.op || loss.Box != c.Op.Intent["target"] || loss.HolderIncarnation != v.holder.Incarnation { |
| 180 | return refused("the loss is not this op's target's under this holder") |
| 181 | } |
| 182 | if fenced, pending := hostclaim.PendingMoveFenceLoss(c); pending { |
| 183 | ref, _ := hostclaim.ConsumedMoveTargetFence(c) |
| 184 | if loss.ID != fenced.ID || loss.Automatic == nil || ref.Key != v.rec.Move.Rollback.LossRef { |
| 185 | return refused("the loss is not the pending fence the record cites") |
| 186 | } |
| 187 | return nil |
| 188 | } |
| 189 | id, err := hostclaim.OperationLossID(c) |
| 190 | if err != nil { |
| 191 | return err |
| 192 | } |
| 193 | if id == "" || hostclaim.OperationLossConsumed(c, id) || loss.ID != id { |
| 194 | return refused("the loss is not the op's pending declaration") |
| 195 | } |
| 196 | if key, err := hostclaim.LossRecordKey(id); err != nil || key != v.rec.Move.Rollback.LossRef { |
| 197 | return refused("the record does not cite the pending declaration") |
| 198 | } |
| 199 | return nil |
| 200 | } |
| 201 | |
| 202 | // frozenMaximum checks the admission's restore point: the recovery's source |
| 203 | // baseline (src_gen, as the first admission selects it, and the recorded |
| 204 | // one once recorded), whose selected files of its own generation reach its |
| 205 | // maximum (genselect.GenerationMax over the plan, never a listing). A plan |
| 206 | // that disagrees waits. |
| 207 | func (r movesRecoveryOwners) frozenMaximum(v recoveryProbe, baseline *genselect.Plan) error { |
| 208 | if baseline == nil || baseline.Generation != selectedGeneration(v.srcGen) { |
| 209 | return fmt.Errorf("%w: the restore point is not the source baseline", hostbox.ErrChanged) |
| 210 | } |
| 211 | recorded, ok, err := r.recordedSourceBaseline(v.rec.Host, v) |
| 212 | if err != nil { |
| 213 | return err |
| 214 | } |
| 215 | if ok && (recorded.Generation != baseline.Generation || recorded.MaxTXID != baseline.MaxTXID) { |
| 216 | return fmt.Errorf("%w: the restore point is not the recorded source baseline", hostbox.ErrChanged) |
| 217 | } |
| 218 | max, err := genselect.GenerationMax(baseline, v.holder.Location, baseline.Generation) |
| 219 | if err != nil { |
| 220 | return fmt.Errorf("%w: the source baseline's plan: %v", hostbox.ErrChanged, err) |
| 221 | } |
| 222 | if max != baseline.MaxTXID { |
| 223 | return fmt.Errorf("%w: the source baseline's files reach %d, not the selection's %d", hostbox.ErrChanged, max, baseline.MaxTXID) |
| 224 | } |
| 225 | return nil |
| 226 | } |
| 227 | |
| 228 | // sourceMaximum vouches for one frozen source (see the file comment) and |
| 229 | // returns its maximum. |
| 230 | func (r movesRecoveryOwners) sourceMaximum(c *hostclaim.Claim, v recoveryProbe, cut uint64, baseline *genselect.Plan, s hostclaim.WriteSource) (uint64, error) { |
| 231 | m := r.m |
| 232 | refused := func(why string) (uint64, error) { |
| 233 | return 0, fmt.Errorf("%w: frozen source %s/%s/%d: %s", hostclaim.ErrLossRecord, s.Box, s.Generation, s.Session, why) |
| 234 | } |
| 235 | kind, region, identity, err := hostops.ParseMoveLossBaseline(s.Baseline) |
| 236 | if err != nil { |
| 237 | return refused("the baseline does not parse") |
| 238 | } |
| 239 | switch kind { |
| 240 | case "replica": |
| 241 | if s.Reason != "target" || s.Box != c.Op.Intent["target"] || s.Generation == "" || identity != s.Generation || s.Session <= cut { |
| 242 | return refused("not a target session since the cut") |
| 243 | } |
| 244 | if !park.SameRegion(v.homeRegion, region, s.Region) || !park.SameRegion(v.homeRegion, s.Region, c.Op.Intent["to"]) { |
| 245 | return refused("not the target region's replica") |
| 246 | } |
| 247 | // The highest TXID of the tuple's own generation the restore |
| 248 | // contains: 0 for a generation outside it. |
| 249 | max, err := genselect.GenerationMax(baseline, v.holder.Location, s.Generation) |
| 250 | if err != nil { |
| 251 | return 0, fmt.Errorf("%w: the restore plan: %v", hostbox.ErrChanged, err) |
| 252 | } |
| 253 | return max, nil |
| 254 | case "seal": |
| 255 | if s.Reason != "source-cut" || identity != v.op || s.Box != m.Box || selectedGeneration(s.Generation) != selectedGeneration(v.srcGen) || s.Session != cut || |
| 256 | !park.SameRegion(v.homeRegion, s.Region, m.Region) { |
| 257 | return refused("not the source's cut session") |
| 258 | } |
| 259 | seal, err := journalledSeal(c) |
| 260 | if err != nil { |
| 261 | return 0, fmt.Errorf("%w: the cut seal: %v", hostbox.ErrChanged, err) |
| 262 | } |
| 263 | if seal.SrcCovered { |
| 264 | return refused("the cut seal covered the source's replica") |
| 265 | } |
| 266 | return seal.N, nil |
| 267 | } |
| 268 | return refused("an unknown baseline") |
| 269 | } |
| 270 | |
| 271 | // lossEvidence is AdmissionEvidence's loss contract: the admission is the |
| 272 | // writer's substitution of this observation's validated sources, which are |
| 273 | // validated again. It consumes the validation. |
| 274 | func (r movesRecoveryOwners) lossEvidence(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) error { |
| 275 | m := r.m |
| 276 | m.mu.Lock() |
| 277 | kept, ok := m.lossValidated[o.Host] |
| 278 | delete(m.lossValidated, o.Host) |
| 279 | m.mu.Unlock() |
| 280 | if !ok || kept.etag != o.ETag { |
| 281 | return fmt.Errorf("%w: no loss sources were validated for this observation", hostbox.ErrChanged) |
| 282 | } |
| 283 | valid, err := r.movesLossSources(ctx, o, kept.record, baseline) |
| 284 | if err != nil { |
| 285 | return err |
| 286 | } |
| 287 | if len(valid) != len(kept.valid) || (len(valid) != 0 && !reflect.DeepEqual(valid, kept.valid)) { |
| 288 | return fmt.Errorf("%w: the loss sources changed", hostbox.ErrChanged) |
| 289 | } |
| 290 | // The writer's substitution (writer_loss.go): the predecessor's |
| 291 | // validated maximum, the rest as additional declarations; a fence |
| 292 | // admits its one frozen target writer and nothing more. |
| 293 | h := o.Claim.Holder |
| 294 | p := hostclaim.AdmissionRecord{Box: h.WriterBox, Session: h.Session, Generation: h.Generation} |
| 295 | if a := h.Admissions; len(a) > 0 { |
| 296 | p = a[len(a)-1] |
| 297 | } |
| 298 | max := baseline.MaxTXID |
| 299 | var more []hostclaim.WritesLost |
| 300 | if _, fenced := hostclaim.PendingMoveFenceLoss(o.Claim); fenced { |
| 301 | if len(valid) != 1 { |
| 302 | return fmt.Errorf("%w: a fence freezes one target writer", hostclaim.ErrFence) |
| 303 | } |
| 304 | max = valid[0].RemoteMaxTXID |
| 305 | } else { |
| 306 | for _, s := range valid { |
| 307 | if s.Box == p.Box && s.Session == p.Session && s.Generation == p.Generation { |
| 308 | max = s.RemoteMaxTXID |
| 309 | } else { |
| 310 | more = append(more, hostclaim.WritesLost{Box: s.Box, Generation: s.Generation, Session: s.Session, RemoteMaxTXID: s.RemoteMaxTXID}) |
| 311 | } |
| 312 | } |
| 313 | } |
| 314 | if in.RemoteMaxTXID != max || len(in.AdditionalWritesLost) != len(more) || (len(more) != 0 && !reflect.DeepEqual(in.AdditionalWritesLost, more)) { |
| 315 | return fmt.Errorf("%w: the admission is not the validated loss sources' substitution", hostbox.ErrChanged) |
| 316 | } |
| 317 | return nil |
| 318 | } |
| 319 | |
amadan