| 1 | // The seeds sweep (seeded-instances spec §3.1, §5.6): the console's |
| 2 | // cleanup of abandoned seed state. It abandons, by a guarded CAS, a seed |
| 3 | // stuck where no box will ever move it; it finishes a rolled-back seed; |
| 4 | // and it deletes a snapshot under seeds/ that no live seed needs. It runs |
| 5 | // unattended and deletes customer data, so every deletion rests on |
| 6 | // positive knowledge: two complete listings, taken in the order that makes |
| 7 | // "no record names it" exact, and a record set with no gap in it. |
| 8 | |
| 9 | package console |
| 10 | |
| 11 | import ( |
| 12 | "context" |
| 13 | "errors" |
| 14 | "log" |
| 15 | "os" |
| 16 | "sort" |
| 17 | "time" |
| 18 | |
| 19 | "github.com/carlosframework/platform/internal/provision" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | "github.com/carlosframework/platform/internal/seeds" |
| 22 | ) |
| 23 | |
| 24 | const ( |
| 25 | seedsPassInterval = 5 * time.Minute |
| 26 | // seedReserveTTL is how long a --from seed may stay reserving. The |
| 27 | // door's copy is bounded at about 52 minutes (store walk, a 30-minute |
| 28 | // restore, 60s, a 20-minute upload, the publish), so an hour is past |
| 29 | // any live copy. That bound leaves out the private-store path's wait |
| 30 | // for its store's walk lock (storeWalks), which has no deadline and |
| 31 | // queues behind holders of up to storeTeardownTimeout each, this |
| 32 | // sweep's own finishes among them. A copy that overruns the hour that |
| 33 | // way is aborted by a CAS with one winner: the user retries, and no |
| 34 | // data is lost. |
| 35 | seedReserveTTL = time.Hour |
| 36 | // seedImportTTL is how long an import may wait for its upload, or for |
| 37 | // the app to say its process is stopped. |
| 38 | seedImportTTL = 7 * 24 * time.Hour |
| 39 | // seedOrphanAge is how old a snapshot no record names must be before |
| 40 | // it is deleted. |
| 41 | seedOrphanAge = time.Hour |
| 42 | ) |
| 43 | |
| 44 | // seedsPassTimeout bounds one pass, so a hung list, read or delete cannot |
| 45 | // stall the sweep for good (the S3 store sets no client timeout). It is |
| 46 | // three intervals: enough for the two listings and a dozen finishes that |
| 47 | // each run to storeTeardownTimeout (60s), the slowest step. A pass cut |
| 48 | // short is resumed by the next tick. A variable so a test can shorten it. |
| 49 | var seedsPassTimeout = 15 * time.Minute |
| 50 | |
| 51 | // The tests' seams, nil in production. seedsPassBetweenListings runs after |
| 52 | // the snapshot listing and before the record listing. |
| 53 | // seedsPassBeforeAbort runs inside each abandonment's mutation, after the |
| 54 | // record is read and before it is written, with the call's number (1, then |
| 55 | // 2 on SetRecordSeed's retry). |
| 56 | var ( |
| 57 | seedsPassBetweenListings func() |
| 58 | seedsPassBeforeAbort func(call int) |
| 59 | ) |
| 60 | |
| 61 | // StartSeedsPass runs the seeds sweep now and every five minutes, until |
| 62 | // ctx is done. A console with host claims armed runs none: its record |
| 63 | // writes go through the claim, every seed door refuses it, and finishSeed |
| 64 | // refuses it too. |
| 65 | func (s *Server) StartSeedsPass(ctx context.Context) { |
| 66 | if s.cfg.HostClaim { |
| 67 | log.Printf("console: seeds pass: off on a host-claim console, where every seed door refuses") |
| 68 | return |
| 69 | } |
| 70 | if s.cfg.Releases == nil || s.cfg.SeedData == nil { |
| 71 | return |
| 72 | } |
| 73 | if _, ok := s.cfg.SeedData.(release.ModLister); !ok { |
| 74 | log.Printf("console: seeds pass: the data bucket cannot list last-modified times, so this sweep deletes no unreferenced snapshot") |
| 75 | } |
| 76 | t := time.NewTicker(seedsPassInterval) |
| 77 | defer t.Stop() |
| 78 | for { |
| 79 | s.seedsPass(ctx, time.Now()) |
| 80 | select { |
| 81 | case <-ctx.Done(): |
| 82 | return |
| 83 | case <-t.C: |
| 84 | } |
| 85 | } |
| 86 | } |
| 87 | |
| 88 | // seedStale reports a seed stuck where no box moves it: a --from seed |
| 89 | // reserving for over seedReserveTTL, or an import waiting for its upload or |
| 90 | // its stop for over seedImportTTL. Both times are a console's: At is |
| 91 | // stamped by the console whose door reserved or began the seed, now is the |
| 92 | // sweeping console's. With one console per deployment they are one clock; |
| 93 | // with more (region HA) their skew comes off the reservation's margin of |
| 94 | // about 8 minutes, and an overrun costs a copy, never data. |
| 95 | func seedStale(seed *provision.SeedSource, now time.Time) bool { |
| 96 | switch { |
| 97 | case seed == nil: |
| 98 | return false |
| 99 | case seed.Kind == seeds.KindReplica && seed.State == seeds.StateReserving: |
| 100 | return now.Sub(seed.At) > seedReserveTTL |
| 101 | case seed.Import() && (seed.State == seeds.StateAwaitingUpload || seed.State == seeds.StateUploaded): |
| 102 | return now.Sub(seed.At) > seedImportTTL |
| 103 | } |
| 104 | return false |
| 105 | } |
| 106 | |
| 107 | // seedsPass is one sweep. In order: |
| 108 | // |
| 109 | // 1. The snapshots under seeds/, with their last-modified times. Listed |
| 110 | // first: every door writes its record before its snapshot (spec §4 |
| 111 | // step 7, §5.1), so a snapshot in this listing whose seed is live has |
| 112 | // its record in the next one. A seed begun after this listing has no |
| 113 | // snapshot in it, so nothing below can reach it. |
| 114 | // 2. Every record, strictly: from the bucket, or, with the record view |
| 115 | // armed and step 1 empty (so nothing can be deleted), from the view. |
| 116 | // Any listing error, here or in step 1, ends the pass: nothing is |
| 117 | // decided on a partial set. A host with a record that would not |
| 118 | // decode is unreadable; a host two records hold is ambiguous. Nothing |
| 119 | // of either is touched: no snapshot deleted, no seed abandoned or |
| 120 | // finished. |
| 121 | // 3. The decisions, one per snapshot, from the two listings alone. |
| 122 | // 4. Each stale seed abandoned by a CAS from the exact state read |
| 123 | // (provision.SeedAbortStale); the box rolls back its part. |
| 124 | // 5. The snapshots deleted: those decided in step 3, and each abandoned |
| 125 | // seed's own once its CAS has landed. |
| 126 | // 6. Each rolled-back seed finished (finishSeed re-reads its record). |
| 127 | // |
| 128 | // Every delete names one exact seeds/<host>/<id>/snapshot.db, so it never |
| 129 | // reaches another seed's. |
| 130 | // |
| 131 | // The pass is bounded by seedsPassTimeout. Cutting it anywhere is safe and |
| 132 | // the next pass resumes: a listing cut short is an error, so nothing is |
| 133 | // decided; an abandonment's CAS either landed (its snapshot then goes next |
| 134 | // pass, as an aborted seed's) or did not (the seed is still stale next |
| 135 | // pass); a snapshot not yet deleted is decided again from fresh listings; |
| 136 | // a finish cut short leaves its record rolled-back, and every finish step |
| 137 | // is idempotent. |
| 138 | func (s *Server) seedsPass(ctx context.Context, now time.Time) { |
| 139 | if s.cfg.HostClaim || s.cfg.Releases == nil || s.cfg.SeedData == nil { |
| 140 | return |
| 141 | } |
| 142 | ctx, cancel := context.WithTimeout(ctx, seedsPassTimeout) |
| 143 | defer cancel() |
| 144 | defer func() { |
| 145 | if ctx.Err() != nil { |
| 146 | log.Printf("console: seeds pass: cut short: %v (the next pass resumes)", ctx.Err()) |
| 147 | } |
| 148 | }() |
| 149 | var snaps []release.KeyMod |
| 150 | ml, walk := s.cfg.SeedData.(release.ModLister) |
| 151 | if walk { |
| 152 | var err error |
| 153 | if snaps, err = ml.ListMod(ctx, seeds.Prefix); err != nil { |
| 154 | log.Printf("console: seeds pass: list snapshots: %v (nothing swept this pass)", err) |
| 155 | return |
| 156 | } |
| 157 | } |
| 158 | if seedsPassBetweenListings != nil { |
| 159 | seedsPassBetweenListings() |
| 160 | } |
| 161 | // Step 2's source. A snapshot listing with nothing in it gives the pass |
| 162 | // nothing to delete, so with the record view armed and able to vouch the |
| 163 | // records come from it (S3 change-index design §4.6): abandonments and |
| 164 | // finishes each act by a CAS on, or a re-read of, the bucket's record. |
| 165 | // With any snapshot listed the walk is the bucket's, strict, after the |
| 166 | // snapshot listing, because a delete is proved by a record's absence. |
| 167 | src := s.cfg.Releases |
| 168 | if len(snaps) == 0 { |
| 169 | src, _ = s.recordWalkStore(ctx, s.cfg.Releases) |
| 170 | } |
| 171 | recs, unreadable, err := provision.ListAllRecordsStrict(ctx, src) |
| 172 | if err != nil { |
| 173 | log.Printf("console: seeds pass: list records: %v (nothing swept this pass)", err) |
| 174 | return |
| 175 | } |
| 176 | byHost := map[string][]provision.Record{} |
| 177 | for _, rec := range recs { |
| 178 | byHost[rec.Host] = append(byHost[rec.Host], rec) |
| 179 | } |
| 180 | // held reports a host the pass must not touch, logging why once. |
| 181 | logged := map[string]bool{} |
| 182 | held := func(host string) bool { |
| 183 | switch { |
| 184 | case unreadable[host]: |
| 185 | if !logged[host] { |
| 186 | log.Printf("console: seeds pass: a record of %s could not be read; nothing of %s is swept this pass", host, host) |
| 187 | } |
| 188 | case len(byHost[host]) > 1: |
| 189 | if !logged[host] { |
| 190 | log.Printf("console: seeds pass: OPERATOR: %d records hold %s; nothing of %s is swept until one remains", len(byHost[host]), host, host) |
| 191 | } |
| 192 | default: |
| 193 | return false |
| 194 | } |
| 195 | logged[host] = true |
| 196 | return true |
| 197 | } |
| 198 | |
| 199 | // Step 3: the decisions. |
| 200 | var condemned []string |
| 201 | waiting := map[[2]string]string{} // host, id -> key: deleted only if this pass abandons the seed |
| 202 | for _, o := range snaps { |
| 203 | host, id, ok := seeds.ParseSnapshotKey(o.Key) |
| 204 | if !ok || held(host) { |
| 205 | continue |
| 206 | } |
| 207 | var rec *provision.Record |
| 208 | if rs := byHost[host]; len(rs) == 1 { |
| 209 | rec = &rs[0] |
| 210 | } |
| 211 | switch snapshotDecision(rec, id, now, o.Mod) { |
| 212 | case snapshotDelete: |
| 213 | condemned = append(condemned, o.Key) |
| 214 | case snapshotIfAbandoned: |
| 215 | waiting[[2]string{host, id}] = o.Key |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | // Step 4: the abandonments. |
| 220 | for _, rec := range recs { |
| 221 | if ctx.Err() != nil { |
| 222 | return |
| 223 | } |
| 224 | if !seedStale(rec.Seed, now) || held(rec.Host) { |
| 225 | continue |
| 226 | } |
| 227 | if s.abandonSeed(ctx, rec) { |
| 228 | if key, ok := waiting[[2]string{rec.Host, rec.Seed.ID}]; ok { |
| 229 | condemned = append(condemned, key) |
| 230 | } |
| 231 | } |
| 232 | } |
| 233 | |
| 234 | // Step 5: the snapshot deletes. |
| 235 | sort.Strings(condemned) |
| 236 | for _, key := range condemned { |
| 237 | if ctx.Err() != nil { |
| 238 | return |
| 239 | } |
| 240 | if err := s.cfg.SeedData.Delete(ctx, key); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 241 | log.Printf("console: seeds pass: delete %s: %v", key, err) |
| 242 | continue |
| 243 | } |
| 244 | log.Printf("console: seeds pass: deleted %s", key) |
| 245 | } |
| 246 | |
| 247 | // Step 6: the finishes. |
| 248 | for _, rec := range recs { |
| 249 | if ctx.Err() != nil { |
| 250 | return |
| 251 | } |
| 252 | if rec.Seed == nil || rec.Seed.State != seeds.StateRolledBack || held(rec.Host) { |
| 253 | continue |
| 254 | } |
| 255 | if err := s.finishSeed(ctx, rec.Account, rec.App, rec.Host, rec.Seed.ID); err != nil { |
| 256 | log.Printf("console: seeds pass: finish seed %s of %s: %v", rec.Seed.ID, rec.Host, err) |
| 257 | } |
| 258 | } |
| 259 | } |
| 260 | |
| 261 | type snapshotVerdict int |
| 262 | |
| 263 | const ( |
| 264 | snapshotKeep snapshotVerdict = iota |
| 265 | snapshotDelete |
| 266 | snapshotIfAbandoned // keep, unless this pass abandons the seed by CAS |
| 267 | ) |
| 268 | |
| 269 | // snapshotDecision decides one snapshot of seed id from the one record that |
| 270 | // holds its host (nil for none), and the snapshot's last-modified time. Its |
| 271 | // age is the console's clock less the bucket's stamp, so skew between |
| 272 | // the two moves it; it only ever decides a snapshot no record names, which |
| 273 | // the listing order has already shown is no live seed's, so the hour is a |
| 274 | // margin and not the proof. |
| 275 | // |
| 276 | // - The record names id: deleted when the seed is placed or committed |
| 277 | // (the host's own replica exists, so the box's delete was lost) or |
| 278 | // aborted or rolled-back (terminal: nothing places it, and the box's |
| 279 | // own rollback deletes the same key; this also covers a trashed app's |
| 280 | // aborted seed, which no box finishes). A seed stale at now waits on |
| 281 | // its abandonment. Any other state is a seed still wanted: kept. |
| 282 | // - The record names another seed, placed or committed: that seed |
| 283 | // superseded this one (spec §3.1). |
| 284 | // - Otherwise no record names it (spec §3.1): deleted once it is over |
| 285 | // seedOrphanAge old. |
| 286 | func snapshotDecision(rec *provision.Record, id string, now, mod time.Time) snapshotVerdict { |
| 287 | if rec != nil && rec.Seed != nil { |
| 288 | state := rec.Seed.State |
| 289 | placed := state == seeds.StatePlaced || state == seeds.StateCommitted |
| 290 | if rec.Seed.ID == id { |
| 291 | switch { |
| 292 | case placed || rec.Seed.Abandoned(): |
| 293 | return snapshotDelete |
| 294 | case seedStale(rec.Seed, now): |
| 295 | return snapshotIfAbandoned |
| 296 | } |
| 297 | return snapshotKeep |
| 298 | } |
| 299 | if placed { |
| 300 | return snapshotDelete |
| 301 | } |
| 302 | } |
| 303 | if now.Sub(mod) > seedOrphanAge { |
| 304 | return snapshotDelete |
| 305 | } |
| 306 | return snapshotKeep |
| 307 | } |
| 308 | |
| 309 | // abandonSeed moves rec's stale seed to aborted by a CAS from exactly the |
| 310 | // state the pass read, and answers whether that CAS landed. A seed that |
| 311 | // moved on meanwhile (the door published, the upload landed, the app said |
| 312 | // stopped) is left alone. |
| 313 | func (s *Server) abandonSeed(ctx context.Context, rec provision.Record) bool { |
| 314 | seed := *rec.Seed |
| 315 | abort := provision.SeedAbortStale(seed.ID, seed.State, seed.At) |
| 316 | call := 0 |
| 317 | _, err := provision.SetRecordSeed(ctx, s.cfg.Releases, rec.Account, rec.App, rec.Host, func(r *provision.Record) error { |
| 318 | call++ |
| 319 | if seedsPassBeforeAbort != nil { |
| 320 | seedsPassBeforeAbort(call) |
| 321 | } |
| 322 | return abort(r) |
| 323 | }) |
| 324 | if err != nil { |
| 325 | if !errors.Is(err, provision.ErrSeedConflict) { |
| 326 | log.Printf("console: seeds pass: abandon seed %s of %s: %v", seed.ID, rec.Host, err) |
| 327 | } |
| 328 | return false |
| 329 | } |
| 330 | log.Printf("console: seeds pass: abandoned seed %s of %s/%s/%s (%s since %s)", seed.ID, rec.Account, rec.App, rec.Host, |
| 331 | seed.State, seed.At.UTC().Format(time.RFC3339)) |
| 332 | s.bumpSerial(ctx) |
| 333 | return true |
| 334 | } |
| 335 | |
amadan