| 1 | // snapshot.go is the accelerator under openDeck: a deck's whole |
| 2 | // document written down every so often, so opening it replays a short |
| 3 | // tail instead of every update since it was made. |
| 4 | // |
| 5 | // The rule the whole file is built on, and the one to keep if anything |
| 6 | // here is ever rewritten: THE LOG IS THE DECK. A snapshot is derived |
| 7 | // data (models.go's DeckSnapshot, Docs' snapshots ruling). Every path |
| 8 | // in this file is therefore allowed to fail without failing the request |
| 9 | // that reached it, and openDeck falls back to the plain log when it |
| 10 | // does. |
| 11 | package slides |
| 12 | |
| 13 | import ( |
| 14 | "context" |
| 15 | "errors" |
| 16 | "fmt" |
| 17 | "time" |
| 18 | |
| 19 | "gorm.io/gorm" |
| 20 | |
| 21 | "amadan.net/oficina/slides/internal/slides/core" |
| 22 | ) |
| 23 | |
| 24 | // snapshotCadence is how many updates may accumulate before the next |
| 25 | // snapshot is taken: the request whose seq lands on a multiple of this |
| 26 | // writes one (maybeSnapshot). |
| 27 | // |
| 28 | // 200 is the plan's number, and the tail it leaves — at most 199 |
| 29 | // updates to replay — sits about twenty times clear of |
| 30 | // core.MaxRetainedUpdates (4096), which is the hard wall a handle hits |
| 31 | // when it has replayed too much to rebuild itself. That headroom is |
| 32 | // conditional on snapshot writes actually succeeding; each failure is |
| 33 | // logged at Error for exactly that reason. |
| 34 | const snapshotCadence = 200 |
| 35 | |
| 36 | const replayBatchSize = 256 |
| 37 | const replayCheckpointCadence = core.MaxRetainedUpdates / 2 |
| 38 | |
| 39 | // latestSnapshot returns the newest usable snapshot for deckID, or nil |
| 40 | // when there is none. |
| 41 | // |
| 42 | // Ordering is seq DESC, id DESC: two writers crossing the boundary |
| 43 | // together may legitimately write equivalent rows at the same seq, so |
| 44 | // the tie resolves to the last row written rather than being treated as |
| 45 | // an error worth reporting. |
| 46 | func (a *app) latestSnapshot(ctx context.Context, deckID int64) (*DeckSnapshot, error) { |
| 47 | var row DeckSnapshot |
| 48 | err := a.db.G.WithContext(ctx). |
| 49 | Where("deck_id = ?", deckID). |
| 50 | Order("seq DESC, id DESC"). |
| 51 | Take(&row).Error |
| 52 | if errors.Is(err, gorm.ErrRecordNotFound) { |
| 53 | return nil, nil |
| 54 | } |
| 55 | if err != nil { |
| 56 | return nil, fmt.Errorf("slides: newest snapshot for deck %d: %w", deckID, err) |
| 57 | } |
| 58 | return &row, nil |
| 59 | } |
| 60 | |
| 61 | // replayPlan is what openDeck has to apply to reconstitute d: the |
| 62 | // newest eligible snapshot (or nil) and the first bounded update batch. |
| 63 | // |
| 64 | // It is one function rather than two calls at the call site so that the |
| 65 | // invariant tying them together — the tail starts exactly where the |
| 66 | // snapshot ends — is written once and can be tested on its own |
| 67 | // (snapshot_test.go). |
| 68 | // |
| 69 | // A snapshot that cannot be read is not an error here. The log alone |
| 70 | // still answers the question, so the failure is returned to the caller |
| 71 | // as "no snapshot" plus a log line, and the deck opens slowly rather |
| 72 | // than not at all. |
| 73 | func (a *app) replayPlan(ctx context.Context, d Deck) (*DeckSnapshot, []DeckUpdate, error) { |
| 74 | snap, err := a.latestSnapshot(ctx, d.ID) |
| 75 | if err != nil { |
| 76 | a.logger.Error("slides: read a deck's newest snapshot, replaying the plain log instead", |
| 77 | "deck", d.Sqid, "err", err) |
| 78 | snap = nil |
| 79 | } |
| 80 | |
| 81 | if snap != nil && (snap.Seq <= 0 || snap.Seq > d.Seq) { |
| 82 | snap = nil |
| 83 | } |
| 84 | var since int64 |
| 85 | if snap != nil { |
| 86 | since = snap.Seq |
| 87 | } |
| 88 | updates, err := a.replayLog(ctx, d, since) |
| 89 | return snap, updates, err |
| 90 | } |
| 91 | |
| 92 | func (a *app) replayLog(ctx context.Context, d Deck, since int64) ([]DeckUpdate, error) { |
| 93 | var updates []DeckUpdate |
| 94 | if err := a.db.G.WithContext(ctx). |
| 95 | Where("deck_id = ? AND seq > ? AND seq <= ?", d.ID, since, d.Seq). |
| 96 | Order("seq").Limit(replayBatchSize).Find(&updates).Error; err != nil { |
| 97 | return nil, fmt.Errorf("slides: read deck %s log: %w", d.Sqid, err) |
| 98 | } |
| 99 | return updates, nil |
| 100 | } |
| 101 | |
| 102 | // maybeSnapshot writes a snapshot when seq lands on the cadence |
| 103 | // boundary, and does nothing at all otherwise. |
| 104 | // |
| 105 | // Two things it is the caller's job to have true, because nothing here |
| 106 | // can check either: |
| 107 | // |
| 108 | // - h holds the document THROUGH seq and no further. Every caller |
| 109 | // applies its update to h and commits it before calling this, so |
| 110 | // the number on the row is the number in the bytes. |
| 111 | // - The transaction that wrote seq has COMMITTED. A snapshot taken |
| 112 | // from a rolled-back state would be a copy of a deck that never |
| 113 | // existed, and openDeck would then skip the real rows it claims to |
| 114 | // cover. |
| 115 | // |
| 116 | // Every failure is logged and swallowed. This runs after the update the |
| 117 | // person actually made is safely on disk; failing their request because |
| 118 | // the accelerator could not be written would be trading the thing that |
| 119 | // matters for the thing that does not. |
| 120 | func (a *app) maybeSnapshot(ctx context.Context, d Deck, h *core.Handle, seq int64) { |
| 121 | if seq <= 0 || seq%snapshotCadence != 0 { |
| 122 | return |
| 123 | } |
| 124 | |
| 125 | state, err := h.State(ctx) |
| 126 | if err != nil { |
| 127 | a.logger.Error("slides: encode a deck snapshot", "deck", d.Sqid, "seq", seq, "err", err) |
| 128 | return |
| 129 | } |
| 130 | |
| 131 | row := &DeckSnapshot{ |
| 132 | DeckID: d.ID, |
| 133 | Seq: seq, |
| 134 | Bytes: state, |
| 135 | CreatedAt: time.Now().UTC(), |
| 136 | } |
| 137 | err = a.db.G.WithContext(ctx).Transaction(func(tx *gorm.DB) error { |
| 138 | if err := tx.Create(row).Error; err != nil { |
| 139 | return fmt.Errorf("slides: persist a snapshot for deck %s at seq %d: %w", d.Sqid, seq, err) |
| 140 | } |
| 141 | // Superseded snapshots are deleted rather than kept: they are |
| 142 | // reproducible from a log that keeps every row, so this deletes |
| 143 | // no history, and retaining them would cost one whole copy of |
| 144 | // the deck per two hundred updates — the difference between an |
| 145 | // accelerator and a second, fatter archive. |
| 146 | // |
| 147 | // "Superseded" is (seq, id) ordered exactly as latestSnapshot |
| 148 | // reads, so what survives is always what latestSnapshot would |
| 149 | // return, and a row at a HIGHER seq — written by another |
| 150 | // request that got further — is never touched. "Delete |
| 151 | // everything that is not mine" would take that one out and |
| 152 | // quietly put the deck back to replaying updates it had |
| 153 | // already been spared. |
| 154 | if err := tx.Where("deck_id = ? AND (seq < ? OR (seq = ? AND id < ?))", |
| 155 | d.ID, row.Seq, row.Seq, row.ID).Delete(&DeckSnapshot{}).Error; err != nil { |
| 156 | return fmt.Errorf("slides: delete superseded snapshots for deck %s at seq %d: %w", d.Sqid, seq, err) |
| 157 | } |
| 158 | return nil |
| 159 | }) |
| 160 | if err != nil { |
| 161 | a.logger.Error("slides: write a deck snapshot", "deck", d.Sqid, "seq", seq, "err", err) |
| 162 | } |
| 163 | } |
| 164 | |
amadan