| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "log" |
| 9 | "strings" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/hostclaim" |
| 13 | "github.com/carlosframework/platform/internal/lease" |
| 14 | "github.com/carlosframework/platform/internal/park" |
| 15 | "github.com/carlosframework/platform/internal/provision" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | // The installer's park and lease adapters (moves spec §6.4, §7.6), shared by |
| 20 | // the landing and the dirty return: the park engine over this box's own |
| 21 | // bucket for the residue parks, with the claim's view and authorizer |
| 22 | // (MoveParks, armMovesParks), the generation classifier the view shares, and |
| 23 | // the own-location lease the build holds from checking through published. |
| 24 | // The phases are in movesinstall.go, the landing's pass in movesland.go. |
| 25 | // |
| 26 | // Nothing stands in for the claim's view or authorizers (Plan 5 Task 9a, |
| 27 | // Q10): an arm without ParkView and ParkAuthorize is not armed, nor one that |
| 28 | // runs a park.Local without ParkLocalAuthorize, and the tests inject fakes. |
| 29 | // Each arm keeps its own operation check in front of the claim's authorizer |
| 30 | // (M17): movesArmAuthorize for the engine's chores, movesLocalAuthorize for |
| 31 | // park.Local's (M9). |
| 32 | |
| 33 | // movesArmAuthorize is an arm's park authorizer (M17): the arm's own |
| 34 | // operation check, then the claim's authorizer, ParkAuthorize from the |
| 35 | // claim's MoveParks. A missing claim authorizer refuses; nothing stands in. |
| 36 | func movesArmAuthorize(claim, own func(context.Context, hostclaim.Observed, hostclaim.Obligation) error) func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { |
| 37 | return func(ctx context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error { |
| 38 | if claim == nil || own == nil { |
| 39 | return fmt.Errorf("%w: no park authorizer", park.ErrUnsafe) |
| 40 | } |
| 41 | if err := own(ctx, obs, o); err != nil { |
| 42 | return err |
| 43 | } |
| 44 | return claim(ctx, obs, o) |
| 45 | } |
| 46 | } |
| 47 | |
| 48 | // movesLocalAuthorize is an arm's park.Local authorizer (M9): the arm's own |
| 49 | // operation check, then the claim's MoveParks local authorizer |
| 50 | // (ParkLocalAuthorize). A missing claim authorizer refuses; the arm's own |
| 51 | // rule never stands alone. |
| 52 | func movesLocalAuthorize(claim, own func(context.Context, *hostclaim.Claim, hostclaim.Obligation) error) func(context.Context, *hostclaim.Claim, hostclaim.Obligation) error { |
| 53 | return func(ctx context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error { |
| 54 | if claim == nil || own == nil { |
| 55 | return fmt.Errorf("%w: no local park authorizer", park.ErrUnsafe) |
| 56 | } |
| 57 | if err := own(ctx, c, o); err != nil { |
| 58 | return err |
| 59 | } |
| 60 | return claim(ctx, c, o) |
| 61 | } |
| 62 | } |
| 63 | |
| 64 | // authorizeMoveChore is an install's own operation check: a chore is run |
| 65 | // only when it is the op's own and the claim still names the op, at the |
| 66 | // install's kind, with this box executing. |
| 67 | func authorizeMoveChore(c *hostclaim.Claim, o hostclaim.Obligation, opID, kind, box string) error { |
| 68 | if c == nil || o.Op != opID || c.Op.ID != opID || c.Op.Kind != kind || c.Op.Exec != box { |
| 69 | return fmt.Errorf("%w: not this install's chore", park.ErrUnsafe) |
| 70 | } |
| 71 | return nil |
| 72 | } |
| 73 | |
| 74 | // authorizeChore is the install's own check for its local chores, which |
| 75 | // park.Local's authorizer runs before the claim's (movesLocalAuthorize, M9). |
| 76 | func (ir *installRun) authorizeChore(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error { |
| 77 | return authorizeMoveChore(c, o, ir.g.ID, ir.src.kind(), ir.cfg().Box) |
| 78 | } |
| 79 | |
| 80 | // authorizeObligation is the install's own operation check over an |
| 81 | // observation, which the engine's authorizer runs before the claim's. |
| 82 | func (ir *installRun) authorizeObligation(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error { |
| 83 | if obs.Absent { |
| 84 | return park.ErrUnsafe |
| 85 | } |
| 86 | return authorizeMoveChore(obs.Claim, o, ir.g.ID, ir.src.kind(), ir.cfg().Box) |
| 87 | } |
| 88 | |
| 89 | // authorizeObligation is the landing's own operation check, the installer's |
| 90 | // rule for a landing. |
| 91 | func (l *landingRun) authorizeObligation(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error { |
| 92 | if obs.Absent { |
| 93 | return park.ErrUnsafe |
| 94 | } |
| 95 | return authorizeMoveChore(obs.Claim, o, l.g.ID, hostclaim.KindLanding, l.cfg().Box) |
| 96 | } |
| 97 | |
| 98 | // landingMarkerGeneration is what landingGenerationOf answers for the |
| 99 | // location's generation.json. The marker is the one key every generation |
| 100 | // reuses (§5.1): the probe's admission supersedes an earlier stay's marker |
| 101 | // with If-Match rather than deleting it, and a park-generation chore whose |
| 102 | // deletion set named the marker key could never be acknowledged once a new |
| 103 | // marker stands there (the engine skips the marker key and then refuses a |
| 104 | // recorded delete at it). So the marker belongs to no generation's chore: |
| 105 | // parking never queues one for it and the engine passes it over. "@" keeps |
| 106 | // it apart from any generation name, as hostclaim.RootGeneration does. |
| 107 | const landingMarkerGeneration = "@marker" |
| 108 | |
| 109 | // landingGenerationOf classifies an object under a holder location by |
| 110 | // generation, as the view does: `g-<gen>/...` and `files/g-<gen>/...` are |
| 111 | // gen's; a level directory (four hex digits) at the location is the |
| 112 | // origin, ""; generation.json is landingMarkerGeneration; anything else is |
| 113 | // unknown, and a caller fails closed on it. The naming rule is |
| 114 | // moves.GenerationPrefix's twin. |
| 115 | func landingGenerationOf(location, key string) (string, bool) { |
| 116 | rel, ok := strings.CutPrefix(key, location+"/") |
| 117 | if !ok || rel == "" { |
| 118 | return "", false |
| 119 | } |
| 120 | for _, p := range []string{"g-", "files/g-"} { |
| 121 | if rest, ok := strings.CutPrefix(rel, p); ok { |
| 122 | if gen, _, ok := strings.Cut(rest, "/"); ok && gen != "" { |
| 123 | return gen, true |
| 124 | } |
| 125 | return "", false |
| 126 | } |
| 127 | } |
| 128 | if rel == "generation.json" { |
| 129 | return landingMarkerGeneration, true |
| 130 | } |
| 131 | if first, _, ok := strings.Cut(rel, "/"); ok && len(first) == 4 && isHexString(first) { |
| 132 | return "", true |
| 133 | } |
| 134 | return "", false |
| 135 | } |
| 136 | |
| 137 | // movesHolderBuildState reports whether key is the build state beside the |
| 138 | // lease at the live holder location (release.BuildStateKey; ruling |
| 139 | // R-PARK): the marker's twin, live state that belongs to no generation. |
| 140 | // A caller passes it over exactly where it passes over the marker at that |
| 141 | // location, before classifying (landingGenerationOf stays strict). The |
| 142 | // same file name anywhere else (a retired location, a moves lane, under a |
| 143 | // generation) is not it. TestMovesHolderBuildStateMatchesTheEngine keeps |
| 144 | // it in step with the park engine. |
| 145 | func movesHolderBuildState(location, key string) bool { |
| 146 | return key == release.BuildStateKey(strings.TrimSuffix(location, "/")) |
| 147 | } |
| 148 | |
| 149 | func isHexString(s string) bool { |
| 150 | for _, c := range s { |
| 151 | if (c < '0' || c > '9') && (c < 'a' || c > 'f') && (c < 'A' || c > 'F') { |
| 152 | return false |
| 153 | } |
| 154 | } |
| 155 | return s != "" |
| 156 | } |
| 157 | |
| 158 | // engine is the park engine over this box's own bucket with the claim's |
| 159 | // leases and view, and the claim's authorizer behind this install's own |
| 160 | // check (the arming requires all three). |
| 161 | func (ir *installRun) engine(vs lease.VersionedStore) park.Engine { |
| 162 | cfg := ir.cfg() |
| 163 | return park.Engine{HomeRegion: cfg.ParkHomeRegion, Store: vs, Claims: ir.s, Bucket: cfg.OwnBucket, Region: cfg.parkRegion(), Leases: cfg.ParkLeases, |
| 164 | Now: cfg.Now, NewID: obligationID, View: cfg.ParkView, Authorize: movesArmAuthorize(cfg.ParkAuthorize, ir.authorizeObligation), After: cfg.parkFault} |
| 165 | } |
| 166 | |
| 167 | // lease is the own-location lease the build holds (§7.6 checking: "hold it |
| 168 | // through published"), kept in the worker's per-host map across passes and |
| 169 | // re-acquired after a restart. waiting is true when another holder has it. |
| 170 | func (ir *installRun) lease(ctx context.Context, holder hostclaim.Holder) (held park.HeldLease, waiting bool, err error) { |
| 171 | return acquireOwnLease(ctx, ir.w, ir.rec.Host, holder) |
| 172 | } |
| 173 | |
| 174 | // acquireOwnLease takes host's own-location lease in this box's own bucket, |
| 175 | // or answers the one w already holds while it still checks. A landing takes |
| 176 | // it at checking (spec §7.6) and keeps it through published; its residue |
| 177 | // parks run under it (ParkRunHeld, M5). waiting is true when another holder |
| 178 | // has it. |
| 179 | func acquireOwnLease(ctx context.Context, w *movesWorker, host string, holder hostclaim.Holder) (held park.HeldLease, waiting bool, err error) { |
| 180 | cfg := w.cfg |
| 181 | w.mu.Lock() |
| 182 | held = w.landLeases[host] |
| 183 | w.mu.Unlock() |
| 184 | if held != nil { |
| 185 | if err := held.Check(ctx); err == nil { |
| 186 | return held, false, nil |
| 187 | } |
| 188 | releaseOwnLease(w, host) |
| 189 | } |
| 190 | held, err = cfg.OwnLease(ctx, cfg.OwnBucket, strings.TrimSuffix(holder.Location, "/")) |
| 191 | if err != nil { |
| 192 | var he *lease.HeldError |
| 193 | if errors.As(err, &he) { |
| 194 | log.Printf("moves: %s: another holder has the location's lease; the build waits", host) |
| 195 | return nil, true, nil |
| 196 | } |
| 197 | return nil, false, fmt.Errorf("the location's lease: %w", err) |
| 198 | } |
| 199 | w.mu.Lock() |
| 200 | if w.landLeases == nil { |
| 201 | w.landLeases = map[string]park.HeldLease{} |
| 202 | } |
| 203 | w.landLeases[host] = held |
| 204 | w.mu.Unlock() |
| 205 | return held, false, nil |
| 206 | } |
| 207 | |
| 208 | // releaseLease releases the host's own-location lease, if held. |
| 209 | func (ir *installRun) releaseLease() { releaseOwnLease(ir.w, ir.rec.Host) } |
| 210 | |
| 211 | // releaseOwnLease releases host's own-location lease on w, if held. The |
| 212 | // landing's pass settles it at its end (settleLease); a return's caller |
| 213 | // does the same once its install call returns; a rollback that abandons |
| 214 | // the landing releases it when the decision takes the record and again on |
| 215 | // every entry to the target's half (rollbackTarget). |
| 216 | func releaseOwnLease(w *movesWorker, host string) { |
| 217 | w.mu.Lock() |
| 218 | held := w.landLeases[host] |
| 219 | delete(w.landLeases, host) |
| 220 | w.mu.Unlock() |
| 221 | if held == nil { |
| 222 | return |
| 223 | } |
| 224 | rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 225 | defer cancel() |
| 226 | if err := held.Release(rctx); err != nil { |
| 227 | log.Printf("moves: %s: release the location's lease: %v", host, err) |
| 228 | } |
| 229 | } |
| 230 | |
| 231 | // settleLease releases the lease once the record is past published, marked |
| 232 | // for a rollback, or no longer landing: the lease is the build's alone. |
| 233 | func (l *landingRun) settleLease() { |
| 234 | m := l.rec.Move |
| 235 | if m == nil || m.State != provision.MoveLanding || m.Reason != "" || !landingBefore(m.LandingPhase, provision.LandLocal) { |
| 236 | releaseOwnLease(l.w, l.rec.Host) |
| 237 | } |
| 238 | } |
| 239 | |
| 240 | // leaseContext derives the context every generation request runs under from |
| 241 | // the lease's own (fix round 2): hostbox's renewingHeld, the production |
| 242 | // park.HeldLease, cancels it when the lease is lost or released, so a |
| 243 | // request in flight, and the S3 client's retries of it, stop with the lease |
| 244 | // rather than landing after it. A lease without a Context method (a plain |
| 245 | // fake) leaves ctx as it is; leasedStore's check before each request still |
| 246 | // stands. |
| 247 | func leaseContext(ctx context.Context, held park.HeldLease) context.Context { |
| 248 | if bound, ok := held.(interface { |
| 249 | Context(context.Context) context.Context |
| 250 | }); ok { |
| 251 | return bound.Context(ctx) |
| 252 | } |
| 253 | return ctx |
| 254 | } |
| 255 | |
| 256 | // leasedStore is the own bucket by version under live lease authority: check |
| 257 | // (the held lease's Check) runs before every request, so a generation write |
| 258 | // after the lease was lost is refused before it is sent, as the cut's |
| 259 | // guarded staging writes are (fix round 1, item 3); the requests themselves |
| 260 | // run under leaseContext, so one already in flight when the lease is lost is |
| 261 | // cancelled with its retries (fix round 2). The build's reads run under both |
| 262 | // too: a holder that has lost the lease should not read ahead either. |
| 263 | type leasedStore struct { |
| 264 | lease.VersionedStore |
| 265 | check func(context.Context) error |
| 266 | } |
| 267 | |
| 268 | func (s leasedStore) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) { |
| 269 | if err := s.check(ctx); err != nil { |
| 270 | return nil, fmt.Errorf("the location's lease: %w", err) |
| 271 | } |
| 272 | return s.VersionedStore.ListVersions(ctx, bucket, prefix) |
| 273 | } |
| 274 | |
| 275 | func (s leasedStore) OpenVersion(ctx context.Context, bucket, key, versionID string) (io.ReadCloser, error) { |
| 276 | if err := s.check(ctx); err != nil { |
| 277 | return nil, fmt.Errorf("the location's lease: %w", err) |
| 278 | } |
| 279 | return s.VersionedStore.OpenVersion(ctx, bucket, key, versionID) |
| 280 | } |
| 281 | |
| 282 | func (s leasedStore) GetVersion(ctx context.Context, bucket, key, versionID string) ([]byte, error) { |
| 283 | if err := s.check(ctx); err != nil { |
| 284 | return nil, fmt.Errorf("the location's lease: %w", err) |
| 285 | } |
| 286 | return s.VersionedStore.GetVersion(ctx, bucket, key, versionID) |
| 287 | } |
| 288 | |
| 289 | func (s leasedStore) CopyVersion(ctx context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) { |
| 290 | if err := s.check(ctx); err != nil { |
| 291 | return "", "", fmt.Errorf("the location's lease: %w", err) |
| 292 | } |
| 293 | return s.VersionedStore.CopyVersion(ctx, srcBucket, srcKey, srcVersionID, dstBucket, dstKey, createOnly) |
| 294 | } |
| 295 | |
| 296 | func (s leasedStore) DeleteVersion(ctx context.Context, bucket, key, versionID string) error { |
| 297 | if err := s.check(ctx); err != nil { |
| 298 | return fmt.Errorf("the location's lease: %w", err) |
| 299 | } |
| 300 | return s.VersionedStore.DeleteVersion(ctx, bucket, key, versionID) |
| 301 | } |
| 302 | |
| 303 | func (s leasedStore) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) { |
| 304 | if err := s.check(ctx); err != nil { |
| 305 | return "", fmt.Errorf("the location's lease: %w", err) |
| 306 | } |
| 307 | return s.VersionedStore.PutIfAbsent(ctx, bucket, key, body) |
| 308 | } |
| 309 | |
amadan