| 1 | package hostlife |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "hash/fnv" |
| 6 | "regexp" |
| 7 | "sort" |
| 8 | "strings" |
| 9 | ) |
| 10 | |
| 11 | // The database and its replica, with the pinned litestream's LTX |
| 12 | // behaviour (rev 7.6, R4-1, R5-1 to R5-3; the facts in |
| 13 | // internal/sim/litestreamfacts). |
| 14 | // |
| 15 | // A database is a map of keys to values with two counters: the lineage |
| 16 | // transaction id (TX), the oracle's order of acknowledged writes, and the |
| 17 | // replication position (Pos), litestream's TXID, which counts shipped LTX |
| 18 | // files and is what every LTX range is numbered in. A replica object is an |
| 19 | // LTX file: a level (0 and 1, 9 for a snapshot), a range in replication |
| 20 | // numbering, the final page image of every key it touches, and, for the |
| 21 | // oracle only, the writes it covers. Pinned litestream writes no database |
| 22 | // checksum into any file (R5-2): a file carries a CRC-64 of itself, which |
| 23 | // the model's files never fail, and nothing else; restore verifies by |
| 24 | // decoding, not by a state checksum. |
| 25 | // |
| 26 | // Replication (R5-1): each shipped file carries one TXID, Pos+1; a |
| 27 | // database with position 0 ships a full-page level-0 file [1, 1] first, |
| 28 | // as litestream's first sync does. Litestream takes a fresh box's position |
| 29 | // from the newest remote level-0 file, and from nothing else: a sidecar is |
| 30 | // follow-mode only, and a snapshot does not count. So a cold restore |
| 31 | // writes a level-0 anchor [cur, cur] with every page before replication |
| 32 | // starts, and the position is cur; without it the position is the newest |
| 33 | // remote level-0 file's, or 0, and replication restarts below the |
| 34 | // snapshot, where the planner never looks. |
| 35 | // |
| 36 | // Rotation and retention (R5-3): the executor's litestream compacts three |
| 37 | // level-0 files into one level-1 file (deleting the inputs), writes a new |
| 38 | // snapshot after each compaction, keeps only the newest snapshot, and |
| 39 | // expires level-1 and level-0 files whose range ends below the previous |
| 40 | // snapshot's TXID. A restore roots at the newest current snapshot of the |
| 41 | // generation at or past the pinned one, and falls back to the pinned |
| 42 | // version only when no successor is current. |
| 43 | |
| 44 | type tx struct { |
| 45 | W int // the global write id, which is what the application acknowledged |
| 46 | TX int // the lineage's transaction id; 0 if the write never reached a database |
| 47 | Key string // the key written |
| 48 | Val int // the value; 0 is a delete |
| 49 | } |
| 50 | |
| 51 | // sum is a digest of the state, for the model's own digests (stop |
| 52 | // evidence, the settle check); no replica file carries it (R5-2). |
| 53 | func sum(state map[string]int) string { |
| 54 | keys := make([]string, 0, len(state)) |
| 55 | for k, v := range state { |
| 56 | if v != 0 { |
| 57 | keys = append(keys, k) |
| 58 | } |
| 59 | } |
| 60 | sort.Strings(keys) |
| 61 | h := fnv.New64a() |
| 62 | for _, k := range keys { |
| 63 | fmt.Fprintf(h, "%s=%d;", k, state[k]) |
| 64 | } |
| 65 | return fmt.Sprintf("%016x", h.Sum64()) |
| 66 | } |
| 67 | |
| 68 | func cloneState(m map[string]int) map[string]int { |
| 69 | out := make(map[string]int, len(m)) |
| 70 | for k, v := range m { |
| 71 | if v != 0 { |
| 72 | out[k] = v |
| 73 | } |
| 74 | } |
| 75 | return out |
| 76 | } |
| 77 | |
| 78 | func statesEqual(a, b map[string]int) (string, bool) { |
| 79 | for k, v := range a { |
| 80 | if v != 0 && b[k] != v { |
| 81 | return fmt.Sprintf("key %s expected %d, got %d", k, v, b[k]), false |
| 82 | } |
| 83 | } |
| 84 | for k, v := range b { |
| 85 | if v != 0 && a[k] != v { |
| 86 | return fmt.Sprintf("key %s expected %d, got %d", k, a[k], v), false |
| 87 | } |
| 88 | } |
| 89 | return "", true |
| 90 | } |
| 91 | |
| 92 | // commit applies acknowledged write w to db as a transaction: mostly an |
| 93 | // insert of a new key named by the write; every fifth write updates the |
| 94 | // newest live key, every fifth-but-one deletes the oldest, so restore |
| 95 | // fidelity depends on final page images, not on a list of ids. It records |
| 96 | // the transaction for the oracle. |
| 97 | func (w *world) commit(db *localDB, wr int) { |
| 98 | if db.State == nil { |
| 99 | db.State = map[string]int{} |
| 100 | } |
| 101 | t := tx{W: wr, TX: db.TX + 1, Key: fmt.Sprintf("k%d", wr), Val: wr} |
| 102 | switch wr % 5 { |
| 103 | case 3: |
| 104 | if k := newestKey(db.State); k != "" { |
| 105 | t.Key, t.Val = k, wr |
| 106 | } |
| 107 | case 4: |
| 108 | if k := oldestKey(db.State); k != "" { |
| 109 | t.Key, t.Val = k, 0 |
| 110 | } |
| 111 | } |
| 112 | if t.Val == 0 { |
| 113 | delete(db.State, t.Key) |
| 114 | } else { |
| 115 | db.State[t.Key] = t.Val |
| 116 | } |
| 117 | db.TX++ |
| 118 | db.Pending = append(db.Pending, wr) |
| 119 | db.Log = append(db.Log, wr) |
| 120 | w.txs[wr] = t |
| 121 | } |
| 122 | |
| 123 | func keyID(k string) int { |
| 124 | n := 0 |
| 125 | fmt.Sscanf(k, "k%d", &n) |
| 126 | return n |
| 127 | } |
| 128 | |
| 129 | func newestKey(state map[string]int) string { |
| 130 | best, id := "", 0 |
| 131 | for k, v := range state { |
| 132 | if v != 0 && keyID(k) > id { |
| 133 | best, id = k, keyID(k) |
| 134 | } |
| 135 | } |
| 136 | return best |
| 137 | } |
| 138 | |
| 139 | func oldestKey(state map[string]int) string { |
| 140 | best, id := "", 0 |
| 141 | for k, v := range state { |
| 142 | if v != 0 && (best == "" || keyID(k) < id) { |
| 143 | best, id = k, keyID(k) |
| 144 | } |
| 145 | } |
| 146 | return best |
| 147 | } |
| 148 | |
| 149 | // ltxKey is the pinned S3 client's layout under a generation prefix: |
| 150 | // <prefix>/<level as four hex digits>/<min>-<max>.ltx, sixteen hex digits |
| 151 | // each (litestream s3/replica_client.go, ltx.FormatFilename). A snapshot |
| 152 | // is level 9 from TXID 1. |
| 153 | func ltxKey(loc, gen string, level, min, max int) string { |
| 154 | return fmt.Sprintf("%s%04x/%016x-%016x.ltx", dataKey(loc, gen, ""), level, min, max) |
| 155 | } |
| 156 | |
| 157 | func snapKey(loc, gen string, max int) string { return ltxKey(loc, gen, 9, 1, max) } |
| 158 | |
| 159 | var ltxKeyRe = regexp.MustCompile(`^(.*/)([0-9a-f]{4})/([0-9a-f]{16})-([0-9a-f]{16})\.ltx$`) |
| 160 | |
| 161 | // ltxLevel parses the level and range from an LTX key; ok is false for |
| 162 | // any other key. |
| 163 | func ltxLevel(k string) (level, min, max int, ok bool) { |
| 164 | m := ltxKeyRe.FindStringSubmatch(k) |
| 165 | if m == nil { |
| 166 | return 0, 0, 0, false |
| 167 | } |
| 168 | fmt.Sscanf(m[2], "%x", &level) |
| 169 | fmt.Sscanf(m[3], "%x", &min) |
| 170 | fmt.Sscanf(m[4], "%x", &max) |
| 171 | return level, min, max, true |
| 172 | } |
| 173 | |
| 174 | // genPrefixOf is the generation prefix an LTX key belongs to: the key |
| 175 | // without its level directory and file name. |
| 176 | func genPrefixOf(k string) string { |
| 177 | m := ltxKeyRe.FindStringSubmatch(k) |
| 178 | if m == nil { |
| 179 | return "" |
| 180 | } |
| 181 | return m[1] |
| 182 | } |
| 183 | |
| 184 | // isLevelDir: a four-hex-digit child is litestream's level directory. |
| 185 | func isLevelDir(child string) bool { |
| 186 | if len(child) != 4 { |
| 187 | return false |
| 188 | } |
| 189 | for _, r := range child { |
| 190 | if !(r >= '0' && r <= '9' || r >= 'a' && r <= 'f') { |
| 191 | return false |
| 192 | } |
| 193 | } |
| 194 | return true |
| 195 | } |
| 196 | |
| 197 | // snapshotOf is the snapshot file of a database at its replication |
| 198 | // position: every page, TXIDs 1 to Pos, and the oracle's list of the |
| 199 | // writes it holds. The caller ships pending writes first, as litestream |
| 200 | // syncs before it snapshots. |
| 201 | // snapshotOf is the snapshot of a database as a seal or an installer |
| 202 | // encodes it: [1, N], N the replication position. A database that holds |
| 203 | // transactions but has never replicated (position 0: a warm database |
| 204 | // whose writes are all pending, a source sealed at its first stay) is |
| 205 | // encoded at TXID 1, the TXID litestream's first sync gives its full-page |
| 206 | // file; only a database with no transaction at all is empty, and the |
| 207 | // encoder refuses a file for it (R8-3; seed 19023 / 30,000 took such a |
| 208 | // seal for empty and lost the source's first write) |
| 209 | func snapshotOf(db *localDB, inc int) *blob { |
| 210 | n := db.Pos |
| 211 | if n == 0 && (len(db.Log) > 0 || len(db.State) > 0) { |
| 212 | n = 1 |
| 213 | } |
| 214 | return &blob{Kind: "snap", Level: 9, Min: 1, Max: n, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: true, |
| 215 | Pages: cloneState(db.State), TXs: append([]int(nil), db.Log...)} |
| 216 | } |
| 217 | |
| 218 | // deltaOf is the next level-0 file: one TXID, Pos+1, with the final |
| 219 | // image of every key the pending writes touched. At position 0 it is the |
| 220 | // first sync, a full-page file [1, 1], as litestream writes it. |
| 221 | func (w *world) deltaOf(db *localDB, inc int) *blob { |
| 222 | pages := map[string]int{} |
| 223 | full := db.Pos == 0 || db.FirstFull |
| 224 | if full { |
| 225 | pages = cloneState(db.State) |
| 226 | } |
| 227 | for _, wr := range db.Pending { |
| 228 | t := w.txs[wr] |
| 229 | pages[t.Key] = db.State[t.Key] // the image now: 0 when deleted since |
| 230 | } |
| 231 | // the oracle's list of the writes the file carries: a full file carries |
| 232 | // the effect of every write applied here, not only the pending ones |
| 233 | txs := db.Pending |
| 234 | if full { |
| 235 | txs = db.Log |
| 236 | } |
| 237 | return &blob{Kind: "ltx", Level: 0, Min: db.Pos + 1, Max: db.Pos + 1, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: full, |
| 238 | Pages: pages, TXs: append([]int(nil), txs...)} |
| 239 | } |
| 240 | |
| 241 | // anchorOf is the level-0 anchor a cold restore writes before replication |
| 242 | // starts (R5-1, the fact in litestreamfacts TestFact_R5_1_AnchorL0ContinuesFromN): |
| 243 | // every page of the restored database as the file [cur, cur], which |
| 244 | // litestream's open takes as its position. |
| 245 | func anchorOf(db *localDB, inc int) *blob { |
| 246 | return &blob{Kind: "ltx", Level: 0, Min: db.Pos, Max: db.Pos, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: true, |
| 247 | Pages: cloneState(db.State), TXs: append([]int(nil), db.Log...)} |
| 248 | } |
| 249 | |
| 250 | // apply overlays an LTX file's page images onto state; a full file replaces |
| 251 | // it, so a key the full file does not carry is gone. |
| 252 | func apply(state map[string]int, f *blob) { |
| 253 | if f.Full { |
| 254 | for k := range state { |
| 255 | if _, ok := f.Pages[k]; !ok { |
| 256 | delete(state, k) |
| 257 | } |
| 258 | } |
| 259 | } |
| 260 | for k, v := range f.Pages { |
| 261 | if v == 0 { |
| 262 | delete(state, k) |
| 263 | } else { |
| 264 | state[k] = v |
| 265 | } |
| 266 | } |
| 267 | } |
| 268 | |
| 269 | // replicate is the serving process's litestream replication (claim C3's |
| 270 | // exception: plain PUTs). A sync writes the pending writes as one local |
| 271 | // level-0 file at Pos+1; ship uploads the local files in order; then |
| 272 | // compaction, snapshot rotation and retention run as the store's |
| 273 | // monitors would. |
| 274 | func (w *world) replicate(b *box) { |
| 275 | if !w.agentUp(b) || b.DB == nil || b.open != nil { |
| 276 | return // replication is not open until the admitted session's start completes (rev 7.11 §7.14) |
| 277 | } |
| 278 | if !b.Old && b.DB.Epoch != b.Epoch { |
| 279 | // the replicator that was admitted died with its agent; a new agent |
| 280 | // process is a new session (claim rev 6.10) and replicates nothing |
| 281 | // for the old one: its wake is a new admission |
| 282 | w.stats["replication refused: the session's agent is gone"]++ |
| 283 | return |
| 284 | } |
| 285 | if !hasMain(b) { |
| 286 | w.closeStep(b) |
| 287 | return |
| 288 | } |
| 289 | w.flush(b) |
| 290 | w.compact(b) |
| 291 | } |
| 292 | |
| 293 | // flush is one sync of the replicator: the pending writes become the next |
| 294 | // local level-0 file, and the queue is shipped. |
| 295 | func (w *world) flush(b *box) { |
| 296 | db := b.DB |
| 297 | if len(db.Pending) > 0 { |
| 298 | // the sync writes this TXID's file locally; the position advances |
| 299 | // at once, and later writes go to the next TXID |
| 300 | bl := w.deltaOf(db, db.Inc) |
| 301 | db.FirstFull = false |
| 302 | db.Queue = append(db.Queue, bl) |
| 303 | db.Pos, db.Pending = bl.Max, nil |
| 304 | } |
| 305 | w.ship(b) |
| 306 | } |
| 307 | |
| 308 | // closedKey is a writer session's clean-close certificate (claim rev 6.8 |
| 309 | // §6.4a): <location>/<generation>/closed/<session>.json, written once and |
| 310 | // never deleted; the admitting CAS moves holder.session, so an old, |
| 311 | // delayed or replayed certificate names a dead session. |
| 312 | func closedKey(loc, gen string, session int) string { |
| 313 | return dataKey(loc, gen, fmt.Sprintf("closed/%d.json", session)) |
| 314 | } |
| 315 | |
| 316 | // isClosedKey: a clean-close certificate of any session. |
| 317 | func isClosedKey(k string) bool { |
| 318 | i := strings.LastIndex(k, "/closed/") |
| 319 | return i >= 0 && strings.HasSuffix(k, ".json") |
| 320 | } |
| 321 | |
| 322 | // track records a request the writer issued whose success the client did |
| 323 | // not see at once, for the clean close's evidence (R8-5, R9-5): every |
| 324 | // class, level-0, compaction, snapshot and retention. A request the server |
| 325 | // refused is a failure the client saw: the session is uncertain for good. |
| 326 | func (db *localDB) track(r *request, v int) { |
| 327 | if v > 0 { |
| 328 | return // client-observed success |
| 329 | } |
| 330 | if r.done { |
| 331 | db.Uncertain = true // refused |
| 332 | return |
| 333 | } |
| 334 | db.Issued = append(db.Issued, r) |
| 335 | } |
| 336 | |
| 337 | // settled: every request the writer issued over its lifetime returned |
| 338 | // success to the client, within the client's wait, and none was retried |
| 339 | // after a timeout (R8-5, R9-5). A request still outstanding within the wait |
| 340 | // is not yet settled; one the client gave up on, whether it later |
| 341 | // completed on the server or not, makes the session uncertain for good, |
| 342 | // since its outcome was never observed. The mutation of close-evidence |
| 343 | // looks at the level-0 queue alone, as rev 7.8 did. |
| 344 | func (db *localDB) settled(now int) bool { |
| 345 | queued := len(db.Queue) == 0 && (db.InFlight == nil || db.InFlight.done) |
| 346 | if !on("close-evidence") { |
| 347 | return queued |
| 348 | } |
| 349 | if !queued || db.Uncertain { |
| 350 | return false |
| 351 | } |
| 352 | for _, r := range db.Issued { |
| 353 | if !r.done { |
| 354 | if now-r.at >= uploadTimeout { |
| 355 | db.Uncertain = true // the client gave up: timeout |
| 356 | return false |
| 357 | } |
| 358 | return false // still waiting |
| 359 | } |
| 360 | if r.doneAt-r.at >= uploadTimeout { |
| 361 | db.Uncertain = true // completed after the client gave up: a lost response |
| 362 | return false |
| 363 | } |
| 364 | } |
| 365 | return true |
| 366 | } |
| 367 | |
| 368 | // closeStep is the clean close of a stopped writer (claim rev 6.8 §6.4a): |
| 369 | // first every request it issued over its lifetime must have returned |
| 370 | // success (R8-5), then it writes closed/<session>.json {session, box, |
| 371 | // max_txid} once (If-None-Match), then it releases its lease. A writer |
| 372 | // with a request unresolved, or that ever retried after a timeout, writes |
| 373 | // no certificate; a crash while draining leaves none (forget). The |
| 374 | // mutation of close-waits writes the certificate as soon as the writer |
| 375 | // stops. |
| 376 | func (w *world) closeStep(b *box) { |
| 377 | db := b.DB |
| 378 | if db == nil || !db.Closing || hasMain(b) { |
| 379 | return |
| 380 | } |
| 381 | if b.open != nil || !b.Old && db.Epoch != b.Epoch { |
| 382 | // the session's start has not opened replication, or its replicator |
| 383 | // died with its agent: nothing certifies (rev 7.11; seed 2663 / |
| 384 | // 30,000 certified position 0 over two unshipped writes) |
| 385 | db.Closing, db.CloseReq = false, nil |
| 386 | w.stats["clean close refused: start unfinished or agent gone"]++ |
| 387 | return |
| 388 | } |
| 389 | if len(b.Procs) > 0 { |
| 390 | // nothing of the host may run when the session certifies: a |
| 391 | // descendant that outlived the unit stop (an escapee outside the |
| 392 | // cgroup) can still acknowledge writes the certificate would not |
| 393 | // cover, which the next admission's purge would lose silently (rev |
| 394 | // 7.10; seed 17 / 300). The close is refused, not delayed: the |
| 395 | // session stays uncertified, the next wake is cold and declares |
| 396 | // (seed 97950 / 100,000: a close that waited for an escapee held |
| 397 | // the revert's probe until it stuck) |
| 398 | db.Closing, db.CloseReq = false, nil |
| 399 | w.stats["clean close refused: a process still runs"]++ |
| 400 | w.logf("box %d: clean close refused, a descendant of the host still runs", b.ID) |
| 401 | return |
| 402 | } |
| 403 | // the certificate is the live generation's: once an operation has |
| 404 | // taken the writer over (a push, a move, a delete), or the claim has |
| 405 | // moved on, the generation is dead or being replaced and the close is |
| 406 | // dropped. A writer running inside an operation (a landing's probe or |
| 407 | // its committed target, a rollback's reopened source, a push's probe) |
| 408 | // certifies like any other (claim rev 6.8 §6.4a; seed 84087 / 100,000: |
| 409 | // rev 7.9's first build dropped every close outside serving, so a |
| 410 | // rollback's source was admitted cold at every wake and the rollback |
| 411 | // never released) |
| 412 | c := w.claim |
| 413 | inside := c != nil && (c.Op.Kind == "serving" || c.Op.Kind == "moving" || c.Op.Kind == "landing" || c.Op.Kind == "restoring" || c.Op.Kind == "reverting") |
| 414 | if c == nil || c.Holder == nil || !inside || c.Op.Exec != b.ID || c.Holder.Inc != db.Inc || db.Gen != c.Holder.Gen { |
| 415 | // a push, a move's cut, a delete or a purge is taking this writer |
| 416 | // over: its generation is being replaced or parked, and a |
| 417 | // certificate written into it now would outlive the park (seed |
| 418 | // 4493 / 30,000: a certificate landed in a generation a push had |
| 419 | // superseded) |
| 420 | db.Closing, db.CloseReq = false, nil |
| 421 | return |
| 422 | } |
| 423 | w.ship(b) |
| 424 | if !db.settled(w.now) && on("close-waits") { |
| 425 | if db.Uncertain && on("close-evidence") { |
| 426 | // a request of this writer timed out and was retried: its outcome |
| 427 | // is unknown for good, so this session never certifies |
| 428 | db.Closing = false |
| 429 | w.stats["clean close refused: uncertain"]++ |
| 430 | w.logf("box %d: clean close refused, the session is uncertain", b.ID) |
| 431 | return |
| 432 | } |
| 433 | inflight := db.InFlight != nil && !db.InFlight.done |
| 434 | w.logf("box %d: clean close waits (queue %d, upload in flight %v, %d requests tracked)", b.ID, len(db.Queue), inflight, len(db.Issued)) |
| 435 | return |
| 436 | } |
| 437 | if db.CloseReq == nil { |
| 438 | db.CloseReq = &request{kind: rPut, region: b.Region, key: closedKey(db.Loc, db.Gen, db.Session), ifSet: true, |
| 439 | b: &blob{Kind: "closed", Gen: db.Gen, Inc: db.Inc, Stamp: db.Stamp, Seq: b.ID, Session: db.Session, TX: db.Pos}} |
| 440 | w.issue(b.ID, db.CloseReq) |
| 441 | w.stats["clean closes"]++ |
| 442 | |
| 443 | } |
| 444 | if db.CloseReq.done { |
| 445 | db.Closing, db.CloseReq = false, nil |
| 446 | } else if w.now-db.CloseReq.at >= uploadTimeout { |
| 447 | // the certificate PUT itself was not acknowledged within the |
| 448 | // client's wait: the close ends without a certificate the box can |
| 449 | // rely on; the PUT may still land, and a later wake that finds it |
| 450 | // may continue warm, since its content is right (the drain was |
| 451 | // complete); meanwhile the next wake is cold (seed 97950 / 100,000: |
| 452 | // a certificate PUT late by three horizons held the revert's probe |
| 453 | // until it stuck) |
| 454 | db.Closing, db.CloseReq = false, nil |
| 455 | w.stats["clean close: certificate not acknowledged"]++ |
| 456 | w.logf("box %d: the certificate PUT was not acknowledged; the close ends uncertified", b.ID) |
| 457 | } |
| 458 | } |
| 459 | |
| 460 | // uploadTimeout is the client's patience with one level-0 upload, in |
| 461 | // steps, before litestream retries it. |
| 462 | const uploadTimeout = 8 |
| 463 | |
| 464 | // ship is the replica's upload loop: the local level-0 files go up in |
| 465 | // TXID order, and the next is never sent before the one ahead of it has |
| 466 | // completed (litestream does not advance past a failed upload, moves spec |
| 467 | // §5.1 step 6). An upload the client has waited uploadTimeout steps for is |
| 468 | // retried with the same bytes to the same key; the original may still land |
| 469 | // later, as an identical version. So a gap in a generation's current |
| 470 | // listing is always a file that has not landed yet, never a file that was |
| 471 | // skipped. |
| 472 | func (w *world) ship(b *box) { |
| 473 | db := b.DB |
| 474 | for len(db.Queue) > 0 { |
| 475 | if r := db.InFlight; r != nil { |
| 476 | if !r.done { |
| 477 | if w.now-db.InFlightAt < uploadTimeout { |
| 478 | return |
| 479 | } |
| 480 | // the client gives up and retries the same bytes; the |
| 481 | // original's outcome is unknown for good (R8-5) |
| 482 | db.Uncertain = true |
| 483 | w.stats["ltx upload retries"]++ |
| 484 | } else { |
| 485 | db.RemotePos = db.Queue[0].Max |
| 486 | db.InFlight = nil |
| 487 | db.Queue = db.Queue[1:] |
| 488 | continue |
| 489 | } |
| 490 | } |
| 491 | bl := db.Queue[0] |
| 492 | // litestream's upload loop (replica.go Sync): the replica reads the |
| 493 | // remote position once, at its first sync, caches it and advances it |
| 494 | // with each upload it completes (R8-5); a local file at or below that |
| 495 | // position is taken as already uploaded and skipped, and the writer |
| 496 | // believes it replicated (R7-1, the initialization/remote-position |
| 497 | // race: another writer's file at this TXID, present at that first |
| 498 | // read, makes this one vanish) |
| 499 | if db.RemotePos < 0 { |
| 500 | db.RemotePos = w.maxL0(w.buckets[b.Region], dataKey(db.Loc, db.Gen, "")) |
| 501 | } |
| 502 | if db.RemotePos >= bl.Max { |
| 503 | w.markDurable(ltxKey(db.Loc, db.Gen, 0, bl.Min, bl.Max), bl) |
| 504 | w.stats["uploads skipped at the remote position"]++ |
| 505 | db.InFlight = nil |
| 506 | db.Queue = db.Queue[1:] |
| 507 | continue |
| 508 | } |
| 509 | r := &request{kind: rPut, region: b.Region, key: ltxKey(db.Loc, db.Gen, 0, bl.Min, bl.Max), b: bl} |
| 510 | if v := w.issue(b.ID, r); v > 0 { |
| 511 | db.RemotePos = bl.Max |
| 512 | db.InFlight = nil |
| 513 | db.Queue = db.Queue[1:] |
| 514 | continue |
| 515 | } |
| 516 | db.track(r, 0) |
| 517 | db.InFlight, db.InFlightAt = r, w.now |
| 518 | return |
| 519 | } |
| 520 | } |
| 521 | |
| 522 | // currentLTX lists the current LTX files of a generation at one level, by |
| 523 | // range. |
| 524 | func (w *world) currentLTX(bk *bucket, prefix string, level int) []*version { |
| 525 | var out []*version |
| 526 | for _, k := range bk.keys(prefix) { |
| 527 | l, _, _, ok := ltxLevel(k) |
| 528 | if !ok || l != level || genPrefixOf(k) != prefix { |
| 529 | continue |
| 530 | } |
| 531 | if v := bk.current(k); v != nil { |
| 532 | out = append(out, v) |
| 533 | } |
| 534 | } |
| 535 | sort.Slice(out, func(i, j int) bool { return out[i].B.Min < out[j].B.Min }) |
| 536 | return out |
| 537 | } |
| 538 | |
| 539 | // compact merges the current level-0 files of b's generation into one |
| 540 | // level-1 file when there are at least three, deletes the inputs bar the |
| 541 | // newest, then |
| 542 | // writes a new snapshot and enforces retention (litestream's store after |
| 543 | // a compaction at the snapshot level): only the newest snapshot stays, |
| 544 | // and level-1 and level-0 files whose range ends below the previous |
| 545 | // snapshot's TXID are deleted. Every delete is a key delete: the versions |
| 546 | // stay noncurrent, which is what keeps the pinned snapshot version |
| 547 | // reachable (R5-3). Contiguity is checked as ltx.Compactor does. |
| 548 | func (w *world) compact(b *box) { |
| 549 | db := b.DB |
| 550 | bk := w.buckets[b.Region] |
| 551 | prefix := dataKey(db.Loc, db.Gen, "") |
| 552 | inputs := w.currentLTX(bk, prefix, 0) |
| 553 | if len(inputs) < 3 { |
| 554 | return |
| 555 | } |
| 556 | out := &blob{Kind: "ltx", Level: 1, Min: inputs[0].B.Min, Inc: db.Inc, Gen: db.Gen, SealKey: db.Key, Pages: map[string]int{}, Full: inputs[0].B.Full} |
| 557 | max := 0 |
| 558 | var keys []string |
| 559 | for _, v := range inputs { |
| 560 | if max != 0 && !(v.B.Min <= max+1 && v.B.Max > max) { |
| 561 | return // non-contiguous inputs: litestream refuses to compact them |
| 562 | } |
| 563 | if v.B.Full { |
| 564 | // a full input anywhere makes the image complete from there: the |
| 565 | // output carries every page and a key the full file lacked is |
| 566 | // gone (rev 7.10: a warm admission's first upload is full, fact |
| 567 | // R5-1c, so full files appear mid-generation; seed 27371 / |
| 568 | // 30,000 compacted one into a delta and a deletion was lost) |
| 569 | out.Full = true |
| 570 | } |
| 571 | apply(out.Pages, v.B) |
| 572 | for k, val := range v.B.Pages { |
| 573 | if val == 0 { |
| 574 | out.Pages[k] = 0 // a delete stays a delete in the compacted image |
| 575 | } |
| 576 | } |
| 577 | out.TXs = append(out.TXs, v.B.TXs...) |
| 578 | if v.B.Max > max { |
| 579 | max = v.B.Max |
| 580 | } |
| 581 | keys = append(keys, ltxKey(db.Loc, db.Gen, 0, v.B.Min, v.B.Max)) |
| 582 | } |
| 583 | out.Max = max |
| 584 | l1 := &request{kind: rPut, region: b.Region, key: ltxKey(db.Loc, db.Gen, 1, out.Min, out.Max), b: out} |
| 585 | if v := w.issue(b.ID, l1); v <= 0 { |
| 586 | db.track(l1, v) // a late or lost compaction upload is part of the writer's close evidence (R8-5) |
| 587 | return |
| 588 | } |
| 589 | // the inputs are deleted bar the newest: litestream's level-0 retention |
| 590 | // never removes the newest level-0 file, which is where a fresh box |
| 591 | // reads its position (facts R5-1f to h; R6 review). The mutation of |
| 592 | // newest-l0 is rev 7.6's: every input, and below every L0 below the |
| 593 | // previous snapshot |
| 594 | keep := keys[len(keys)-1] |
| 595 | for _, k := range keys { |
| 596 | if k == keep && on("newest-l0") { |
| 597 | continue |
| 598 | } |
| 599 | dr := &request{kind: rDelKey, region: b.Region, key: k} |
| 600 | db.track(dr, w.issue(b.ID, dr)) |
| 601 | } |
| 602 | w.stats["ltx compactions"]++ |
| 603 | // snapshot rotation: a new snapshot at the position, then retention |
| 604 | prev := 0 |
| 605 | for _, s := range w.currentLTX(bk, prefix, 9) { |
| 606 | if s.B.Max > prev { |
| 607 | prev = s.B.Max |
| 608 | } |
| 609 | } |
| 610 | snap := snapshotOf(db, db.Inc) |
| 611 | snap.Kind = "ltx" |
| 612 | if snap.Max <= prev { |
| 613 | return |
| 614 | } |
| 615 | sr := &request{kind: rPut, region: b.Region, key: snapKey(db.Loc, db.Gen, snap.Max), b: snap} |
| 616 | if v := w.issue(b.ID, sr); v <= 0 { |
| 617 | db.track(sr, v) |
| 618 | return |
| 619 | } |
| 620 | w.stats["snapshot rotations"]++ |
| 621 | // keep the newest snapshot; minSnapshotTXID is the previous one's |
| 622 | // MaxTXID (litestream db.EnforceSnapshotRetention); expire levels 1 |
| 623 | // and 0 below it |
| 624 | for _, s := range w.currentLTX(bk, prefix, 9) { |
| 625 | if s.B.Max < snap.Max { |
| 626 | dr := &request{kind: rDelKey, region: b.Region, key: snapKey(db.Loc, db.Gen, s.B.Max)} |
| 627 | db.track(dr, w.issue(b.ID, dr)) |
| 628 | } |
| 629 | } |
| 630 | newest := w.maxL0(bk, prefix) |
| 631 | for _, level := range []int{1, 0} { |
| 632 | for _, f := range w.currentLTX(bk, prefix, level) { |
| 633 | if level == 0 && f.B.Max == newest && on("newest-l0") { |
| 634 | continue // the newest level-0 file survives every retention path (R7-3) |
| 635 | } |
| 636 | if f.B.Max < prev { |
| 637 | dr := &request{kind: rDelKey, region: b.Region, key: ltxKey(db.Loc, db.Gen, level, f.B.Min, f.B.Max)} |
| 638 | db.track(dr, w.issue(b.ID, dr)) |
| 639 | } |
| 640 | } |
| 641 | } |
| 642 | } |
| 643 | |
| 644 | // remoteMax is the generation's remote maximum TXID over every level. |
| 645 | func (w *world) remoteMax(bk *bucket, prefix string) int { |
| 646 | max := 0 |
| 647 | for _, level := range []int{0, 1, 9} { |
| 648 | for _, f := range w.currentLTX(bk, prefix, level) { |
| 649 | if f.B.Max > max { |
| 650 | max = f.B.Max |
| 651 | } |
| 652 | } |
| 653 | } |
| 654 | return max |
| 655 | } |
| 656 | |
| 657 | // plan is litestream's restore planner (replica.go CalcRestorePlan, |
| 658 | // v0.5.14) over the current files under prefix, from TXID cur: take the |
| 659 | // file that extends the range, overlap allowed (ltx.IsContiguous: |
| 660 | // min <= cur+1 && max > cur), preferring the highest max, then the lowest |
| 661 | // min, then the higher level; stop when nothing extends it. The mutation |
| 662 | // of restore-overlap is rev 7.4's rule, min == cur+1, which discards a |
| 663 | // compacted continuation. With restore-gap-closed, a current file that |
| 664 | // starts beyond cur+1 once nothing extends the range is a gap and the |
| 665 | // restore refuses (litestream: "non-contiguous ltx files"); the mutation |
| 666 | // returns what it has. |
| 667 | func (w *world) plan(bk *bucket, prefix string, cur int) (files []*blob, gap bool) { |
| 668 | var cands []*blob |
| 669 | for _, k := range bk.keys(prefix) { |
| 670 | level, _, _, ok := ltxLevel(k) |
| 671 | if !ok || level == 9 || genPrefixOf(k) != prefix { |
| 672 | continue // snapshots are the root, never planned |
| 673 | } |
| 674 | if v := bk.current(k); v != nil { |
| 675 | cands = append(cands, v.B) |
| 676 | } |
| 677 | } |
| 678 | for { |
| 679 | var best *blob |
| 680 | for _, f := range cands { |
| 681 | ext := f.Min <= cur+1 && f.Max > cur |
| 682 | if !on("restore-overlap") { |
| 683 | ext = f.Min == cur+1 |
| 684 | } |
| 685 | if !ext { |
| 686 | continue |
| 687 | } |
| 688 | if best == nil || f.Max > best.Max || f.Max == best.Max && (f.Min < best.Min || f.Min == best.Min && f.Level > best.Level) { |
| 689 | best = f |
| 690 | } |
| 691 | } |
| 692 | if best == nil { |
| 693 | break |
| 694 | } |
| 695 | files = append(files, best) |
| 696 | cur = best.Max |
| 697 | } |
| 698 | for _, f := range cands { |
| 699 | if f.Min > cur+1 { |
| 700 | gap = true |
| 701 | } |
| 702 | } |
| 703 | return files, gap |
| 704 | } |
| 705 | |
| 706 | // maxL0 is the newest current level-0 file's TXID of a generation: what |
| 707 | // litestream's open takes as a fresh box's position (R5-1). |
| 708 | func (w *world) maxL0(bk *bucket, prefix string) int { |
| 709 | max := 0 |
| 710 | for _, f := range w.currentLTX(bk, prefix, 0) { |
| 711 | if f.B.Max > max { |
| 712 | max = f.B.Max |
| 713 | } |
| 714 | } |
| 715 | return max |
| 716 | } |
| 717 | |
| 718 | // expectState replays the incarnation's acknowledged, non-exempt writes, |
| 719 | // in acknowledgement order, into the database a correct restore must |
| 720 | // equal. With upto >= 0 it replays the transactions with lineage TXID 1 |
| 721 | // to upto, what a cold restore that reached upto must hold; with upto < |
| 722 | // 0 it replays every acknowledged write, a write that never reached a |
| 723 | // database counting as the insert the application believed it made, which |
| 724 | // is what the serving box's own database must hold. |
| 725 | func (w *world) expectState(inc int, upto int) map[string]int { |
| 726 | state := map[string]int{} |
| 727 | for _, a := range w.acks { |
| 728 | if a.Inc != inc || a.Exempt || a.Kind != "write" { |
| 729 | continue |
| 730 | } |
| 731 | t := w.txs[a.W] |
| 732 | if upto >= 0 && (t.TX == 0 || t.TX > upto) { |
| 733 | continue |
| 734 | } |
| 735 | if t.Key == "" { |
| 736 | t.Key, t.Val = fmt.Sprintf("k%d", a.W), a.W |
| 737 | } |
| 738 | if t.Val == 0 { |
| 739 | delete(state, t.Key) |
| 740 | } else { |
| 741 | state[t.Key] = t.Val |
| 742 | } |
| 743 | } |
| 744 | return state |
| 745 | } |
| 746 | |
| 747 | // txsOf is the union, in order, of the writes the files cover: the oracle's |
| 748 | // log of a restored database. |
| 749 | func txsOf(files ...*blob) []int { |
| 750 | seen := map[int]bool{} |
| 751 | var out []int |
| 752 | for _, f := range files { |
| 753 | for _, wr := range f.TXs { |
| 754 | if !seen[wr] { |
| 755 | seen[wr] = true |
| 756 | out = append(out, wr) |
| 757 | } |
| 758 | } |
| 759 | } |
| 760 | return out |
| 761 | } |
| 762 | |
| 763 | // lineageTX is the oracle's lineage TXID of a restored state: the highest |
| 764 | // transaction id among the writes the applied files cover. |
| 765 | func (w *world) lineageTX(files ...*blob) int { |
| 766 | max := 0 |
| 767 | for _, f := range files { |
| 768 | for _, wr := range f.TXs { |
| 769 | if t := w.txs[wr].TX; t > max { |
| 770 | max = t |
| 771 | } |
| 772 | } |
| 773 | } |
| 774 | return max |
| 775 | } |
| 776 | |
| 777 | var _ = strings.HasPrefix |
| 778 | |
amadan