| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "encoding/hex" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "reflect" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/activator" |
| 16 | "github.com/carlosframework/platform/internal/genselect" |
| 17 | "github.com/carlosframework/platform/internal/hostbox" |
| 18 | "github.com/carlosframework/platform/internal/hostclaim" |
| 19 | "github.com/carlosframework/platform/internal/hostfiles" |
| 20 | "github.com/carlosframework/platform/internal/hostrelocate" |
| 21 | "github.com/carlosframework/platform/internal/hostwrite" |
| 22 | "github.com/carlosframework/platform/internal/moves" |
| 23 | "github.com/carlosframework/platform/internal/park" |
| 24 | "github.com/carlosframework/platform/internal/registry" |
| 25 | "github.com/carlosframework/platform/internal/release" |
| 26 | ) |
| 27 | |
| 28 | // RelocationWakeInputs is the relocating kind's ClaimWakeInputs (claim |
| 29 | // Plan 11 R1). Field names are distinct (finding 14): FilesManifest is |
| 30 | // moves' C8 provider, FilesStore the data bucket as a release.Store. |
| 31 | type RelocationWakeInputs struct { |
| 32 | Launch func(context.Context, registry.Route) (ClaimServingLaunch, error) |
| 33 | FilesManifest ClaimFilesManifest |
| 34 | Markers ClaimMarkerStore |
| 35 | Claims release.Store |
| 36 | FilesStore release.Store |
| 37 | Data genselect.Reader |
| 38 | Bucket string |
| 39 | Region string |
| 40 | // HomeRegion is the box's verified home binding (hostbox.Actor's), or |
| 41 | // "": the selection and the fresh publisher see the holder through it |
| 42 | // (park.HolderRegionView), as main's serving owners do. |
| 43 | HomeRegion string |
| 44 | Box string |
| 45 | NewID func() string |
| 46 | Now func() time.Time |
| 47 | // Releases verifies a journaled release at its receipt (stage A's |
| 48 | // StopReceiptReleases over main's Complete). |
| 49 | Releases hostclaim.ReleaseAuthority |
| 50 | |
| 51 | mu sync.Mutex |
| 52 | fences map[string]hostwrite.Work // host\x00session: the admitted spawn selection Materialize ran under |
| 53 | published map[string]ClaimPublicationReceipt // host\x00session: P's prepared marker |
| 54 | fresh *claimServingOwners // a fresh generation's marker: main's writer (ruling 47) |
| 55 | } |
| 56 | |
| 57 | var ( |
| 58 | _ ClaimWakeInputs = (*RelocationWakeInputs)(nil) |
| 59 | _ ClaimFencedPublication = (*RelocationWakeInputs)(nil) |
| 60 | ) |
| 61 | |
| 62 | func (r *RelocationWakeInputs) valid() bool { |
| 63 | return r != nil && r.Launch != nil && r.FilesManifest != nil && r.Markers != nil && r.Claims != nil && r.FilesStore != nil && r.Data != nil && |
| 64 | r.Bucket != "" && r.Region != "" && r.Box != "" && r.NewID != nil && r.Now != nil && r.Releases != nil |
| 65 | } |
| 66 | |
| 67 | func (r *RelocationWakeInputs) probe(o hostclaim.Observed, p activator.StartPolicy) error { |
| 68 | if !r.valid() { |
| 69 | return hostwrite.ErrUnavailable |
| 70 | } |
| 71 | c := o.Claim |
| 72 | if o.Absent || c == nil || c.Holder == nil || c.Op.Kind != hostclaim.KindRelocating || c.Op.Phase != hostclaim.PhaseRelocatingProbing || c.Op.Exec != r.Box || !p.Probe || c.Op.Journal["stuck"] != "" { |
| 73 | return fmt.Errorf("%w: not this box's relocation probe", hostrelocate.ErrRefused) |
| 74 | } |
| 75 | return nil |
| 76 | } |
| 77 | |
| 78 | // freshOwners is main's serving publisher bound to this box and region, |
| 79 | // used only through its kind-free halves (publishFresh, freshEvidence). |
| 80 | func (r *RelocationWakeInputs) freshOwners() *claimServingOwners { |
| 81 | r.mu.Lock() |
| 82 | defer r.mu.Unlock() |
| 83 | if r.fresh == nil { |
| 84 | r.fresh = &claimServingOwners{HomeRegion: r.HomeRegion, Data: r.Data, Bucket: r.Bucket, Region: r.Region, Box: r.Box, NewID: r.NewID, Now: r.Now, Sources: ClaimServingSources{Markers: r.Markers}} |
| 85 | } |
| 86 | return r.fresh |
| 87 | } |
| 88 | |
| 89 | func (r *RelocationWakeInputs) request(o hostclaim.Observed) genselect.Request { |
| 90 | return genselect.Request{Bucket: r.Bucket, Region: r.Region, Host: o.Host, Holder: park.HolderRegionView(*o.Claim.Holder, r.HomeRegion, r.Region)} |
| 91 | } |
| 92 | |
| 93 | func preparedAuthority(c *hostclaim.Claim) (hostclaim.PreparedAuthority, error) { |
| 94 | var auth hostclaim.PreparedAuthority |
| 95 | if json.Unmarshal([]byte(c.Op.Journal["prepared_authority"]), &auth) != nil || auth.Generation == "" || auth.Generation != c.Op.Journal["prepared"] { |
| 96 | return hostclaim.PreparedAuthority{}, hostbox.ErrChanged |
| 97 | } |
| 98 | return auth, nil |
| 99 | } |
| 100 | |
| 101 | // emptyPrepared reports whether P is the evidenced empty bootstrap: the |
| 102 | // journaled pin says so (RecoveryPin.Empty, which the pinner admits only |
| 103 | // for a release reporting final_txid 0 or a holder with no writer) and the |
| 104 | // prepared authority carries no install manifest. The two must agree: an |
| 105 | // unreadable pin, or a pin and an authority that disagree, is an error, |
| 106 | // never an empty P. A missing install manifest alone is not evidence of |
| 107 | // an empty database, and a later attempt's empty listing is legitimate |
| 108 | // only over an empty P (a pinned snapshot lives in the location, so an |
| 109 | // empty listing under a non-empty pin is a wrong bucket or a listing not |
| 110 | // yet visible). |
| 111 | func emptyPrepared(c *hostclaim.Claim, auth hostclaim.PreparedAuthority) (bool, error) { |
| 112 | var ref hostclaim.RecoveryPin |
| 113 | if json.Unmarshal([]byte(c.Op.Journal["pin"]), &ref) != nil { |
| 114 | return false, fmt.Errorf("%w: the journaled pin is unreadable", hostbox.ErrChanged) |
| 115 | } |
| 116 | if ref.Empty != (auth.InstallManifest == "") { |
| 117 | return false, fmt.Errorf("%w: the pin and the prepared authority disagree on an empty P", hostbox.ErrChanged) |
| 118 | } |
| 119 | return ref.Empty, nil |
| 120 | } |
| 121 | |
| 122 | // Select (§5.8 "The probe's admission"): the first attempt opens P with |
| 123 | // KeepStamp, certified only by this operation's verified release (C4); |
| 124 | // it is a prepared admission (Prepared, main's pre-replication marker) |
| 125 | // when P has a snapshot, and an empty one otherwise (ruling 47). Every |
| 126 | // later attempt is cold into a fresh generation from the predecessor. |
| 127 | func (r *RelocationWakeInputs) Select(ctx context.Context, rt registry.Route, o hostclaim.Observed, p activator.StartPolicy) (ClaimLaunchInputs, error) { |
| 128 | if err := r.probe(o, p); err != nil { |
| 129 | return ClaimLaunchInputs{}, err |
| 130 | } |
| 131 | c, h := o.Claim, o.Claim.Holder |
| 132 | if rt.Host != o.Host || !strings.HasPrefix(rt.DBPath, "/") { |
| 133 | return ClaimLaunchInputs{}, hostbox.ErrUnsupported |
| 134 | } |
| 135 | l, err := r.Launch(ctx, rt) |
| 136 | if err != nil { |
| 137 | return ClaimLaunchInputs{}, err |
| 138 | } |
| 139 | if l.Identity.Config == "" || l.Identity.Release == "" || l.Executable == "" || len(l.ExecutableSHA256) != 64 || l.Version == "" { |
| 140 | return ClaimLaunchInputs{}, hostwrite.ErrUnavailable |
| 141 | } |
| 142 | if l.Version != c.Op.Journal["install_release"] { |
| 143 | return ClaimLaunchInputs{}, fmt.Errorf("%w: the staged release %s is not the pinned %s", hostrelocate.ErrRefused, l.Version, c.Op.Journal["install_release"]) |
| 144 | } |
| 145 | prepared := c.Op.Journal["prepared"] |
| 146 | in := ClaimLaunchInputs{Roles: []string{"main"}, Identity: l.Identity, Executable: l.Executable, ExecutableSHA256: l.ExecutableSHA256, Args: append([]string(nil), l.Args...), Env: append([]string(nil), l.Env...), |
| 147 | DBPath: rt.DBPath, Version: l.Version, Staged: l.Staged, Limit: l.Limit, ReadyPath: "/"} |
| 148 | adm := hostclaim.Admission{ObservedOpID: c.Op.ID, ObservedPhase: hostclaim.PhaseRelocatingProbing, ObservedSession: h.Session, ObservedGeneration: h.Generation, |
| 149 | Box: r.Box, Mode: "cold", Region: r.Region, PredecessorRegion: r.Region} |
| 150 | if n := len(h.Admissions); n > 0 { |
| 151 | adm.PredecessorRegion = h.Admissions[n-1].Region |
| 152 | } |
| 153 | if !hostclaim.PreparedOpened(c) { |
| 154 | certified, err := hostrelocate.CertifiedFor(c) |
| 155 | if err != nil { |
| 156 | return ClaimLaunchInputs{}, err |
| 157 | } |
| 158 | if certified { |
| 159 | var rel hostclaim.Release |
| 160 | if err := json.Unmarshal([]byte(c.Op.Journal["release"]), &rel); err != nil { |
| 161 | return ClaimLaunchInputs{}, err |
| 162 | } |
| 163 | if err := r.Releases.VerifyRelease(ctx, *c, rel); err != nil { |
| 164 | return ClaimLaunchInputs{}, err |
| 165 | } |
| 166 | } |
| 167 | auth, err := preparedAuthority(c) |
| 168 | if err != nil { |
| 169 | return ClaimLaunchInputs{}, err |
| 170 | } |
| 171 | empty, err := emptyPrepared(c, auth) |
| 172 | if err != nil { |
| 173 | return ClaimLaunchInputs{}, err |
| 174 | } |
| 175 | adm.NewGeneration, adm.KeepStamp, adm.PredecessorCertified = prepared, true, certified |
| 176 | if empty { |
| 177 | // An evidenced empty pin opened as P: no snapshot, so no root |
| 178 | // until the first transaction; main's fenced publisher waits |
| 179 | // for it (ErrClaimRootPending). |
| 180 | in.Empty = true |
| 181 | in.Admission = adm |
| 182 | return in, nil |
| 183 | } |
| 184 | im, _, err := readInstallManifest(ctx, r.Data, r.Bucket, moves.GenerationPrefix(h.Location, prepared)) |
| 185 | if err != nil { |
| 186 | return ClaimLaunchInputs{}, err |
| 187 | } |
| 188 | plan, err := genselect.SelectPrepared(ctx, r.Data, r.request(o), genselect.PreparedPin{Generation: prepared, SnapshotKey: im.SnapshotKey, SnapshotVersion: im.SnapshotVersion, SnapshotTXID: im.N, SnapshotCRC64: im.SnapshotCRC64}) |
| 189 | if err != nil { |
| 190 | return ClaimLaunchInputs{}, err |
| 191 | } |
| 192 | in.Prepared = true |
| 193 | adm.RemoteMaxTXID, in.Baseline = plan.MaxTXID, plan |
| 194 | in.Admission = adm |
| 195 | return in, nil |
| 196 | } |
| 197 | adm.NewGeneration = r.NewID() |
| 198 | if !claimFreshGeneration(h, adm.NewGeneration) { |
| 199 | return ClaimLaunchInputs{}, hostwrite.ErrUnavailable |
| 200 | } |
| 201 | plan, err := genselect.Select(ctx, r.Data, r.request(o)) |
| 202 | if errors.Is(err, genselect.ErrEmpty) { |
| 203 | // A later attempt follows P's admitted session, which ran: a |
| 204 | // listing that finds nothing is absence (nothing replicated, or |
| 205 | // not yet visible), never an empty predecessor. Main's serving |
| 206 | // rule admits an empty cold admission only for a holder that never |
| 207 | // admitted a writer (claimEmptyBootstrap), which a later attempt |
| 208 | // never is (fix round 1, I2). |
| 209 | return ClaimLaunchInputs{}, fmt.Errorf("%w: the location lists nothing after P's admitted session", hostbox.ErrChanged) |
| 210 | } |
| 211 | if err != nil { |
| 212 | return ClaimLaunchInputs{}, err |
| 213 | } |
| 214 | adm.RemoteMaxTXID, in.Baseline = plan.MaxTXID, plan |
| 215 | in.Admission = adm |
| 216 | return in, nil |
| 217 | } |
| 218 | |
| 219 | // Materialize, under the admitted spawn selection: the held identities |
| 220 | // proved, C8 establishes the files manifest (Establish publishes |
| 221 | // <host>.env's exact bytes before it returns, O1), the selection is |
| 222 | // recorded as the session's fence, and on P's prepared opening P's marker |
| 223 | // is published under that fence (main's prepared-publication contract, |
| 224 | // ruling 47). The held render and release ran at the local install. |
| 225 | func (r *RelocationWakeInputs) Materialize(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, work hostwrite.Work) error { |
| 226 | if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil { |
| 227 | return err |
| 228 | } |
| 229 | if work == nil || in.Identity.Config == "" || in.Identity.Release == "" { |
| 230 | return hostwrite.ErrUnavailable |
| 231 | } |
| 232 | fence := writerFence(work, in.Identity) |
| 233 | if err := fence(ctx); err != nil { |
| 234 | return err |
| 235 | } |
| 236 | if err := r.FilesManifest.Establish(ctx, o, in, work); err != nil { |
| 237 | return err |
| 238 | } |
| 239 | h := o.Claim.Holder |
| 240 | r.mu.Lock() |
| 241 | if r.fences == nil { |
| 242 | r.fences = map[string]hostwrite.Work{} |
| 243 | } |
| 244 | r.fences[claimPublishedKey(o.Host, h.Session)] = work |
| 245 | r.mu.Unlock() |
| 246 | if !in.Prepared { |
| 247 | return nil |
| 248 | } |
| 249 | receipt, err := r.preparedMarker(ctx, o, fence) |
| 250 | if err != nil { |
| 251 | return err |
| 252 | } |
| 253 | r.record(o.Host, h.Session, receipt) |
| 254 | return nil |
| 255 | } |
| 256 | |
| 257 | func (r *RelocationWakeInputs) record(host string, session uint64, receipt ClaimPublicationReceipt) { |
| 258 | r.mu.Lock() |
| 259 | defer r.mu.Unlock() |
| 260 | if r.published == nil { |
| 261 | r.published = map[string]ClaimPublicationReceipt{} |
| 262 | } |
| 263 | r.published[claimPublishedKey(host, session)] = receipt |
| 264 | } |
| 265 | |
| 266 | // preparedMarker establishes P's verified marker under fence: install.json |
| 267 | // is the journaled prepared authority's bytes, and the location's |
| 268 | // generation.json names P with the build's pin (left as it is when it |
| 269 | // already does) or supersedes the current marker by its exact ETag. |
| 270 | func (r *RelocationWakeInputs) preparedMarker(ctx context.Context, o hostclaim.Observed, fence func(context.Context) error) (ClaimPublicationReceipt, error) { |
| 271 | c, h := o.Claim, o.Claim.Holder |
| 272 | auth, err := preparedAuthority(c) |
| 273 | if err != nil { |
| 274 | return ClaimPublicationReceipt{}, err |
| 275 | } |
| 276 | bounded, cancel := context.WithTimeout(ctx, hostbox.RequestLimit) |
| 277 | defer cancel() |
| 278 | im, raw, err := readInstallManifest(bounded, r.Data, r.Bucket, moves.GenerationPrefix(h.Location, auth.Generation)) |
| 279 | if err != nil { |
| 280 | return ClaimPublicationReceipt{}, err |
| 281 | } |
| 282 | sum := sha256.Sum256(raw) |
| 283 | if hex.EncodeToString(sum[:]) != auth.SHA256 || im.Host != o.Host || im.N != auth.SnapshotTXID || h.Generation != auth.Generation { |
| 284 | return ClaimPublicationReceipt{}, fmt.Errorf("%w: install.json is not the prepared authority's", hostbox.ErrChanged) |
| 285 | } |
| 286 | root := genselect.File{Key: im.SnapshotKey, VersionID: im.SnapshotVersion, Level: 9, MinTXID: 1, MaxTXID: im.N} |
| 287 | current, err := currentMarker(bounded, r.Data, r.Bucket, h) // moves' reader, one implementation |
| 288 | if err != nil { |
| 289 | return ClaimPublicationReceipt{}, err |
| 290 | } |
| 291 | expected := "" |
| 292 | if current != nil { |
| 293 | got, err := r.Data.GetVersion(bounded, r.Bucket, markerKey(h), current.ID) |
| 294 | if err != nil { |
| 295 | return ClaimPublicationReceipt{}, err |
| 296 | } |
| 297 | var marker genselect.Marker |
| 298 | if err := json.Unmarshal(got, &marker); err != nil { |
| 299 | return ClaimPublicationReceipt{}, err |
| 300 | } |
| 301 | if marker.Op == auth.Generation { |
| 302 | if marker.Incarnation != h.Incarnation || marker.Host != o.Host || marker.Box != r.Box || marker.ManifestSHA256 != auth.SHA256 { |
| 303 | return ClaimPublicationReceipt{}, hostbox.ErrChanged |
| 304 | } |
| 305 | return ClaimPublicationReceipt{MarkerVersion: current.ID, Root: root}, nil |
| 306 | } |
| 307 | expected = current.ETag |
| 308 | } |
| 309 | body, err := json.Marshal(genselect.Marker{Op: auth.Generation, Incarnation: h.Incarnation, Host: o.Host, Box: r.Box, InstalledAt: r.Now().UTC(), ManifestSHA256: auth.SHA256, Pin: manifestPin(im)}) |
| 310 | if err != nil { |
| 311 | return ClaimPublicationReceipt{}, err |
| 312 | } |
| 313 | if err := fence(bounded); err != nil { |
| 314 | return ClaimPublicationReceipt{}, err |
| 315 | } |
| 316 | version, err := r.Markers.PutMarker(bounded, r.Bucket, markerKey(h), body, expected, fence) |
| 317 | if err != nil { |
| 318 | return ClaimPublicationReceipt{}, err |
| 319 | } |
| 320 | if version == "" { |
| 321 | return ClaimPublicationReceipt{}, hostwrite.ErrUnavailable |
| 322 | } |
| 323 | return ClaimPublicationReceipt{MarkerVersion: version, Root: root}, nil |
| 324 | } |
| 325 | |
| 326 | // EmptyAuthority: P's opening is empty only with an evidenced empty pin |
| 327 | // (the pinner's RecoveryPin.Empty, which the prepared authority agrees |
| 328 | // with). A later attempt is never empty: it follows P's admitted session, |
| 329 | // and main's serving rule admits an empty cold admission only for a holder |
| 330 | // that never admitted a writer (fix round 1, I2). |
| 331 | func (r *RelocationWakeInputs) EmptyAuthority(_ context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error { |
| 332 | if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil { |
| 333 | return err |
| 334 | } |
| 335 | if !in.Empty || in.Baseline != nil || in.Prepared { |
| 336 | return hostbox.ErrChanged |
| 337 | } |
| 338 | auth, err := preparedAuthority(o.Claim) |
| 339 | if err != nil { |
| 340 | return err |
| 341 | } |
| 342 | empty, err := emptyPrepared(o.Claim, auth) |
| 343 | if err != nil { |
| 344 | return err |
| 345 | } |
| 346 | if in.Admission.NewGeneration != o.Claim.Op.Journal["prepared"] || hostclaim.PreparedOpened(o.Claim) { |
| 347 | return fmt.Errorf("%w: a later attempt follows P's admitted session and is never empty", hostbox.ErrChanged) |
| 348 | } |
| 349 | if !empty { |
| 350 | return fmt.Errorf("%w: no evidenced empty pin", hostrelocate.ErrRefused) |
| 351 | } |
| 352 | return nil |
| 353 | } |
| 354 | |
| 355 | // Publish is main's pre-replication call for a prepared admission |
| 356 | // (ClaimWakeAssembly s.Replicate when in.Prepared). It finds, or writes, |
| 357 | // P's marker under the fence Materialize recorded for the admitted session |
| 358 | // and records the receipt; without that fence it refuses. Any other call |
| 359 | // is refused: a fresh generation is published fenced (PublishFenced). |
| 360 | func (r *RelocationWakeInputs) Publish(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs) error { |
| 361 | if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil { |
| 362 | return err |
| 363 | } |
| 364 | h := o.Claim.Holder |
| 365 | if !in.Prepared || in.Admission.NewGeneration != o.Claim.Op.Journal["prepared"] || h.Generation != in.Admission.NewGeneration { |
| 366 | return hostbox.ErrChanged |
| 367 | } |
| 368 | r.mu.Lock() |
| 369 | work := r.fences[claimPublishedKey(o.Host, h.Session)] |
| 370 | r.mu.Unlock() |
| 371 | if work == nil { |
| 372 | return fmt.Errorf("%w: no admitted writer's fence for session %d", hostbox.ErrChanged, h.Session) |
| 373 | } |
| 374 | receipt, err := r.preparedMarker(ctx, o, writerFence(work, in.Identity)) |
| 375 | if err != nil { |
| 376 | return err |
| 377 | } |
| 378 | r.record(o.Host, h.Session, receipt) |
| 379 | return nil |
| 380 | } |
| 381 | |
| 382 | // PublishFenced is main's fresh-generation marker writer, for a later |
| 383 | // attempt's generation and an empty P's opening (ruling 47). |
| 384 | func (r *RelocationWakeInputs) PublishFenced(ctx context.Context, o hostclaim.Observed, in ClaimLaunchInputs, pub ClaimPublication) (ClaimPublicationReceipt, error) { |
| 385 | if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil { |
| 386 | return ClaimPublicationReceipt{}, err |
| 387 | } |
| 388 | if in.Prepared { |
| 389 | return ClaimPublicationReceipt{}, hostbox.ErrChanged |
| 390 | } |
| 391 | return r.freshOwners().publishFresh(ctx, o, in, pub) |
| 392 | } |
| 393 | |
| 394 | // AdmissionEvidence binds a relocation probe admission to its observation |
| 395 | // and to the certified baseline, read again: on P's opening the exact |
| 396 | // prepared pin and P's files manifest by digest (prepared_authority); on a |
| 397 | // later attempt the predecessor's fresh baseline. |
| 398 | func (r *RelocationWakeInputs) AdmissionEvidence(ctx context.Context, o hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) error { |
| 399 | if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil { |
| 400 | return err |
| 401 | } |
| 402 | c, h := o.Claim, o.Claim.Holder |
| 403 | if in.ObservedOpID != c.Op.ID || in.ObservedPhase != hostclaim.PhaseRelocatingProbing || in.ObservedSession != h.Session || in.ObservedGeneration != h.Generation || in.Box != r.Box || in.Mode != "cold" || in.Coverage != "" || len(in.AdditionalWritesLost) != 0 || in.Region != r.Region { |
| 404 | return hostbox.ErrChanged |
| 405 | } |
| 406 | prepared := c.Op.Journal["prepared"] |
| 407 | auth, err := preparedAuthority(c) |
| 408 | if err != nil { |
| 409 | return err |
| 410 | } |
| 411 | empty, err := emptyPrepared(c, auth) |
| 412 | if err != nil { |
| 413 | return err |
| 414 | } |
| 415 | if in.NewGeneration == prepared { |
| 416 | certified, err := hostrelocate.CertifiedFor(c) |
| 417 | if err != nil { |
| 418 | return err |
| 419 | } |
| 420 | if !in.KeepStamp || hostclaim.PreparedOpened(c) || in.PredecessorCertified != certified { |
| 421 | return hostbox.ErrChanged |
| 422 | } |
| 423 | raw, _, err := r.FilesStore.Get(ctx, hostfiles.ManifestKey(moves.GenerationFilesPrefix(h.Location, prepared))) |
| 424 | if err != nil { |
| 425 | return err |
| 426 | } |
| 427 | if sum := sha256.Sum256(raw); hex.EncodeToString(sum[:]) != auth.FilesManifestSHA256 { |
| 428 | return fmt.Errorf("%w: P's files manifest is not the prepared authority's", hostbox.ErrChanged) |
| 429 | } |
| 430 | if baseline == nil { |
| 431 | if !empty || in.RemoteMaxTXID != 0 { |
| 432 | return hostbox.ErrChanged |
| 433 | } |
| 434 | return nil |
| 435 | } |
| 436 | if baseline.Generation != prepared || baseline.MaxTXID != in.RemoteMaxTXID || baseline.MaxTXID != auth.SnapshotTXID { |
| 437 | return hostbox.ErrChanged |
| 438 | } |
| 439 | return nil |
| 440 | } |
| 441 | if in.KeepStamp || in.PredecessorCertified || !claimFreshGeneration(h, in.NewGeneration) { |
| 442 | return hostbox.ErrChanged |
| 443 | } |
| 444 | if baseline == nil { |
| 445 | // A later attempt follows P's admitted session and is never an |
| 446 | // empty cold admission (fix round 1, I2). |
| 447 | return hostbox.ErrChanged |
| 448 | } |
| 449 | fresh, err := genselect.Select(ctx, r.Data, r.request(o)) |
| 450 | if err != nil { |
| 451 | return err |
| 452 | } |
| 453 | if fresh.Generation != baseline.Generation || fresh.MaxTXID != baseline.MaxTXID || in.RemoteMaxTXID != fresh.MaxTXID || !reflect.DeepEqual(fresh.Files, baseline.Files) { |
| 454 | return hostbox.ErrChanged |
| 455 | } |
| 456 | return nil |
| 457 | } |
| 458 | |
| 459 | // PublicationEvidence: on P's prepared opening, this process's receipt for |
| 460 | // the admitted session with the marker still exactly that version; on any |
| 461 | // fresh generation, main's freshEvidence. |
| 462 | func (r *RelocationWakeInputs) PublicationEvidence(ctx context.Context, o hostclaim.Observed) error { |
| 463 | if err := r.probe(o, activator.StartPolicy{Probe: true}); err != nil { |
| 464 | return err |
| 465 | } |
| 466 | h := o.Claim.Holder |
| 467 | r.mu.Lock() |
| 468 | receipt, prepared := r.published[claimPublishedKey(o.Host, h.Session)] |
| 469 | r.mu.Unlock() |
| 470 | if !prepared { |
| 471 | return r.freshOwners().freshEvidence(ctx, o) |
| 472 | } |
| 473 | current, err := currentMarker(ctx, r.Data, r.Bucket, h) |
| 474 | if err != nil { |
| 475 | return err |
| 476 | } |
| 477 | if current == nil || current.ID != receipt.MarkerVersion || h.Generation != o.Claim.Op.Journal["prepared"] { |
| 478 | return hostbox.ErrChanged |
| 479 | } |
| 480 | return nil |
| 481 | } |
| 482 | |
amadan