| 1 | package docs |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "time" |
| 8 | |
| 9 | "gorm.io/gorm" |
| 10 | |
| 11 | "github.com/carlosframework/docs/internal/core" |
| 12 | ) |
| 13 | |
| 14 | // coreABIVersion is core.ABIVersion, aliased here so this package's own |
| 15 | // snapshot code and tests read as docs code rather than reaching into |
| 16 | // internal/core at every call site. It is the same constant, not a |
| 17 | // second number that could drift from it. |
| 18 | const coreABIVersion = core.ABIVersion |
| 19 | |
| 20 | // snapshotCadence is how many updates may accumulate past a snapshot |
| 21 | // before the next one is taken. 256 keeps a replay to a snapshot plus a |
| 22 | // short tail, which is what holds the retained-update count in a |
| 23 | // core.Handle clear of maxRetainedUpdates (4096) — about 16x of headroom. |
| 24 | // |
| 25 | // That headroom is conditional, and the condition is worth knowing: |
| 26 | // it holds while snapshot writes keep succeeding. Roughly fifteen |
| 27 | // consecutive failures — each logged at Error, so noisy rather than |
| 28 | // silent — walk the tail back to the cap. |
| 29 | const snapshotCadence = 256 |
| 30 | |
| 31 | // newestSnapshot returns the most recent usable snapshot for docID, or |
| 32 | // nil when there is none. |
| 33 | // |
| 34 | // "Usable" excludes rows written by a different core ABI: the state |
| 35 | // encoding is that core's business, and a row from another one is |
| 36 | // discarded rather than trusted. Ordering is seq DESC, id DESC — two |
| 37 | // concurrent POSTs at the cadence boundary may legitimately write |
| 38 | // equivalent snapshots at the same seq, so the tie resolves to the last |
| 39 | // row written rather than being treated as an error. |
| 40 | func (s *Store) newestSnapshot(ctx context.Context, docID int64) (*DocSnapshot, error) { |
| 41 | var row DocSnapshot |
| 42 | err := s.db.WithContext(ctx). |
| 43 | Where("doc_id = ? AND abi_version = ?", docID, coreABIVersion). |
| 44 | Order("seq DESC, id DESC"). |
| 45 | First(&row).Error |
| 46 | if errors.Is(err, gorm.ErrRecordNotFound) { |
| 47 | return nil, nil |
| 48 | } |
| 49 | if err != nil { |
| 50 | return nil, fmt.Errorf("docs: newest snapshot for document %d: %w", docID, err) |
| 51 | } |
| 52 | var doc Document |
| 53 | if err := s.db.WithContext(ctx).Select("guid").First(&doc, docID).Error; err != nil { |
| 54 | return nil, err |
| 55 | } |
| 56 | if err := s.openContentSnapshot(doc.GUID, &row); err != nil { |
| 57 | return nil, err |
| 58 | } |
| 59 | return &row, nil |
| 60 | } |
| 61 | |
| 62 | // writeSnapshot encodes h's current state as one Yjs update, asks the |
| 63 | // core to verify it is this document, and persists it as doc_snapshots |
| 64 | // row (doc.ID, seq) — then deletes every snapshot for the document |
| 65 | // that this one supersedes. |
| 66 | // |
| 67 | // seq must be the highest seq h actually holds, and the caller is |
| 68 | // responsible for that being true: see ApplyUpdate, where this is |
| 69 | // called from the replayed state before the incoming update is |
| 70 | // applied, which is what makes the number truthful (the snapshots |
| 71 | // design, D3). Nothing here can check it. Verification compares the |
| 72 | // candidate against h itself, so a snapshot faithfully copied from a |
| 73 | // state that is missing a committed row passes perfectly — D5 says so |
| 74 | // in as many words. Placement closes that hole, not this function. |
| 75 | // |
| 76 | // Every error is returned rather than logged here, so the one call |
| 77 | // site logs each of them once, at Error, with the same doc_id and seq |
| 78 | // keys: a failed encode, a failed verification and a failed insert all |
| 79 | // mean the same thing to an operator, which is that this document did |
| 80 | // not get the snapshot it was due. None of them may fail the update — |
| 81 | // by the time this matters the log is what the document is, and it is |
| 82 | // untouched. |
| 83 | func (s *Store) writeSnapshot(ctx context.Context, doc *Document, h *core.Handle, seq int64) error { |
| 84 | candidate, err := h.Snapshot(ctx) |
| 85 | if err != nil { |
| 86 | return fmt.Errorf("docs: encode snapshot for document %d at seq %d: %w", doc.ID, seq, err) |
| 87 | } |
| 88 | |
| 89 | // Verified before it is trusted, inside the core (D5): the |
| 90 | // candidate is replayed into a throwaway document there and |
| 91 | // compared by state vector, delete set and structural render. A |
| 92 | // false verdict is a refusal to write, never a warning attached to |
| 93 | // a row that gets written anyway — Store.replay reads whatever is |
| 94 | // on disk back and has no way to second-guess it beyond noticing |
| 95 | // that the core refused the bytes outright. |
| 96 | verified, err := h.VerifySnapshot(ctx, candidate) |
| 97 | if err != nil { |
| 98 | return fmt.Errorf("docs: verify snapshot for document %d at seq %d: %w", doc.ID, seq, err) |
| 99 | } |
| 100 | if !verified { |
| 101 | return fmt.Errorf("docs: snapshot for document %d at seq %d did not verify as this document; not written", doc.ID, seq) |
| 102 | } |
| 103 | |
| 104 | row := &DocSnapshot{ |
| 105 | DocID: doc.ID, |
| 106 | Seq: seq, |
| 107 | State: candidate, |
| 108 | ABIVersion: coreABIVersion, |
| 109 | CreatedAt: time.Now().UTC(), |
| 110 | } |
| 111 | return s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { |
| 112 | if err := s.snapshotPermission(ctx, tx, doc.ID); err != nil { |
| 113 | return err |
| 114 | } |
| 115 | if err := s.prepareContentSnapshot(tx, doc.GUID, row); err != nil { |
| 116 | return err |
| 117 | } |
| 118 | if err := tx.Create(row).Error; err != nil { |
| 119 | return fmt.Errorf("docs: persist snapshot for document %d at seq %d: %w", doc.ID, seq, err) |
| 120 | } |
| 121 | |
| 122 | // Superseded snapshots are derived data, reproducible from a |
| 123 | // log that keeps every row, so deleting them deletes no |
| 124 | // history — and retaining them would cost one whole document |
| 125 | // copy per snapshotCadence updates, which is the difference |
| 126 | // between an accelerator and a second, fatter copy of the |
| 127 | // archive. |
| 128 | // |
| 129 | // "Superseded" is (seq, id) ordered exactly as newestSnapshot |
| 130 | // reads: strictly lower seq, or the same seq and a lower id. |
| 131 | // So what survives is always what newestSnapshot would return, |
| 132 | // and a row this write does not supersede is never touched — |
| 133 | // including one at a HIGHER seq. |
| 134 | // |
| 135 | // That last part is the reason not to write the simpler |
| 136 | // "delete every other row for this document". Two writers can |
| 137 | // cross the boundary together, and the one that commits second |
| 138 | // is not necessarily the one that replayed furthest: a writer |
| 139 | // working from a prefix that stopped at seq 260 can commit its |
| 140 | // snapshot after another has already committed one at seq 300. |
| 141 | // Under this predicate the seq 300 row is neither lower-seq nor |
| 142 | // an equal-seq lower id, so it stays and newestSnapshot keeps |
| 143 | // preferring it. Under "everything that is not mine" it would |
| 144 | // be deleted, and the document would quietly go back to |
| 145 | // replaying forty updates it had already been spared. Rows at |
| 146 | // the same seq are ordinary here rather than a collision to |
| 147 | // resolve — migration 0004 makes (doc_id, seq) non-unique for |
| 148 | // exactly this reason, since two boundary writers snapshotting |
| 149 | // the same prefix produce equivalent rows. |
| 150 | if err := tx.Where("doc_id = ? AND (seq < ? OR (seq = ? AND id < ?))", doc.ID, row.Seq, row.Seq, row.ID). |
| 151 | Delete(&DocSnapshot{}).Error; err != nil { |
| 152 | return fmt.Errorf("docs: delete superseded snapshots for document %d at seq %d: %w", doc.ID, seq, err) |
| 153 | } |
| 154 | return s.snapshotPermission(ctx, tx, doc.ID) |
| 155 | }) |
| 156 | } |
| 157 | |
amadan