pds.dad
This repository has no description
1package internal
2
3import (
4 "bufio"
5 "bytes"
6 "encoding/binary"
7 "fmt"
8 "io"
9 "os"
10 "path/filepath"
11 "strings"
12 "time"
13
14 "github.com/ipfs/go-cid"
15 car "github.com/ipld/go-car"
16)
17
18// CarStore manages on-disk append-only CAR files, one per repo DID.
19//
20// Each file starts life as the exact CAR returned by com.atproto.sync.getRepo.
21// New commit blocks from the firehose are appended to the end of the file and
22// the header root is patched in place to the new commit CID, so the header
23// root always points at the LATEST commit and standard CAR tooling sees the
24// current repo state. Blocks from older commits remain in the file behind the
25// root, preserving history. The current head (rev + commit CID + prevData) is
26// also tracked in the sqlite DB.
27type CarStore struct {
28 dir string
29 trashDir string
30}
31
32func NewCarStore(dataDir string) (*CarStore, error) {
33 dir := filepath.Join(dataDir, "repos")
34 if err := os.MkdirAll(dir, 0o755); err != nil {
35 return nil, fmt.Errorf("failed to create repo store dir: %w", err)
36 }
37 trashDir := filepath.Join(dataDir, "trash")
38 if err := os.MkdirAll(trashDir, 0o755); err != nil {
39 return nil, fmt.Errorf("failed to create trash dir: %w", err)
40 }
41 return &CarStore{dir: dir, trashDir: trashDir}, nil
42}
43
44func (s *CarStore) carPath(did string) string {
45 // DIDs are filesystem-safe (did:plc:xyz); guard against path tricks anyway
46 clean := strings.ReplaceAll(did, string(filepath.Separator), "_")
47 return filepath.Join(s.dir, clean+".car")
48}
49
50func (s *CarStore) trashPath(did string) string {
51 clean := strings.ReplaceAll(did, string(filepath.Separator), "_")
52 return filepath.Join(s.trashDir, clean+".car")
53}
54
55// journalSuffix marks the root-patch journal sidecar next to a CAR file.
56const journalSuffix = ".rootj"
57
58// InitCar writes a full repo CAR to disk atomically (temp file + rename),
59// replacing any existing file for the DID.
60func (s *CarStore) InitCar(did string, carBytes []byte) error {
61 start := time.Now()
62 defer func() { carInitDuration.Observe(time.Since(start).Seconds()) }()
63
64 final := s.carPath(did)
65 tmp := final + ".tmp"
66
67 f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o644)
68 if err != nil {
69 return fmt.Errorf("failed to create temp CAR: %w", err)
70 }
71 if _, err := f.Write(carBytes); err != nil {
72 f.Close()
73 os.Remove(tmp)
74 return fmt.Errorf("failed to write CAR: %w", err)
75 }
76 if err := f.Sync(); err != nil {
77 f.Close()
78 os.Remove(tmp)
79 return fmt.Errorf("failed to fsync CAR: %w", err)
80 }
81 if err := f.Close(); err != nil {
82 os.Remove(tmp)
83 return fmt.Errorf("failed to close CAR: %w", err)
84 }
85 // a stale journal from the file being replaced must never be replayed
86 // over the new CAR
87 if err := os.Remove(final + journalSuffix); err != nil && !os.IsNotExist(err) {
88 os.Remove(tmp)
89 return fmt.Errorf("failed to remove stale root journal: %w", err)
90 }
91 if err := os.Rename(tmp, final); err != nil {
92 os.Remove(tmp)
93 return fmt.Errorf("failed to rename CAR into place: %w", err)
94 }
95 return nil
96}
97
98// AppendCommit appends every block from a commit event's CAR slice to the
99// repo's CAR file, then patches the header root to newHead so the file's
100// declared root is the latest commit. Returns an error if the repo has no
101// CAR file yet.
102func (s *CarStore) AppendCommit(did string, blocks []byte, newHead cid.Cid) (int, error) {
103 start := time.Now()
104 path := s.carPath(did)
105 if _, err := os.Stat(path); err != nil {
106 return 0, fmt.Errorf("no CAR file for repo %s: %w", did, err)
107 }
108
109 // heal a torn header from a crash mid-root-patch before touching the file
110 if err := recoverJournal(path); err != nil {
111 return 0, fmt.Errorf("failed to recover root journal: %w", err)
112 }
113
114 // parse the commit slice
115 cr, err := car.NewCarReader(bytes.NewReader(blocks))
116 if err != nil {
117 return 0, fmt.Errorf("failed to read commit CAR slice: %w", err)
118 }
119
120 // buffer the serialized blocks first so we don't append a partial
121 // commit if the slice is malformed halfway through
122 var buf bytes.Buffer
123 w := bufio.NewWriter(&buf)
124 n := 0
125 for {
126 blk, err := cr.Next()
127 if err == io.EOF {
128 break
129 }
130 if err != nil {
131 return 0, fmt.Errorf("failed to read block from commit slice: %w", err)
132 }
133 cidBytes := blk.Cid().Bytes()
134 data := blk.RawData()
135
136 var prefix [binary.MaxVarintLen64]byte
137 pn := binary.PutUvarint(prefix[:], uint64(len(cidBytes)+len(data)))
138 if _, err := w.Write(prefix[:pn]); err != nil {
139 return 0, err
140 }
141 if _, err := w.Write(cidBytes); err != nil {
142 return 0, err
143 }
144 if _, err := w.Write(data); err != nil {
145 return 0, err
146 }
147 n++
148 }
149 if err := w.Flush(); err != nil {
150 return 0, err
151 }
152
153 f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0o644)
154 if err != nil {
155 return 0, fmt.Errorf("failed to open CAR for append: %w", err)
156 }
157 if _, err := f.Write(buf.Bytes()); err != nil {
158 f.Close()
159 return 0, fmt.Errorf("failed to append blocks: %w", err)
160 }
161 if err := f.Sync(); err != nil {
162 f.Close()
163 return 0, fmt.Errorf("failed to fsync CAR: %w", err)
164 }
165 if err := f.Close(); err != nil {
166 return 0, fmt.Errorf("failed to close CAR: %w", err)
167 }
168
169 // blocks are durable; now flip the header root to the new commit. A crash
170 // before this point just leaves the old root, fixed by the next append.
171 if err := patchRoot(path, newHead); err != nil {
172 return 0, fmt.Errorf("failed to update CAR header root: %w", err)
173 }
174
175 carAppends.Inc()
176 carAppendBytes.Add(float64(len(blocks)))
177 carAppendDuration.Observe(time.Since(start).Seconds())
178 return n, nil
179}
180
181// patchRoot rewrites the single root CID in a CAR file's header in place.
182// The new CID must serialize to the same length as the existing root (always
183// true for atproto commits: CIDv1/dag-cbor/sha2-256, 36 bytes), so the
184// varint-length-prefixed header does not change size.
185//
186// The full patched header region is journaled to a sidecar file before the
187// in-place write, so a crash mid-write leaves a journal that recoverJournal
188// can replay instead of a torn header.
189func patchRoot(path string, newHead cid.Cid) error {
190 f, err := os.OpenFile(path, os.O_RDWR, 0o644)
191 if err != nil {
192 return fmt.Errorf("failed to open CAR: %w", err)
193 }
194 defer f.Close()
195
196 region, header, varintLen, err := readHeaderRegion(f)
197 if err != nil {
198 return err
199 }
200
201 h, err := car.ReadHeader(bufio.NewReader(bytes.NewReader(region)))
202 if err != nil {
203 return fmt.Errorf("failed to parse CAR header: %w", err)
204 }
205 if len(h.Roots) != 1 {
206 return fmt.Errorf("expected exactly one root in CAR header, got %d", len(h.Roots))
207 }
208
209 oldRoot := h.Roots[0]
210 if oldRoot.Equals(newHead) {
211 return nil
212 }
213
214 oldBytes := oldRoot.Bytes()
215 newBytes := newHead.Bytes()
216 if len(oldBytes) != len(newBytes) {
217 return fmt.Errorf("new root CID length %d does not match existing root length %d", len(newBytes), len(oldBytes))
218 }
219
220 off := bytes.Index(header, oldBytes)
221 if off < 0 {
222 return fmt.Errorf("existing root CID not found in CAR header bytes")
223 }
224 copy(region[varintLen+off:], newBytes)
225
226 if err := writeFileAtomic(path+journalSuffix, region); err != nil {
227 return fmt.Errorf("failed to write root journal: %w", err)
228 }
229
230 if _, err := f.WriteAt(newBytes, int64(varintLen+off)); err != nil {
231 return fmt.Errorf("failed to write new root CID: %w", err)
232 }
233 if err := f.Sync(); err != nil {
234 return fmt.Errorf("failed to fsync CAR header: %w", err)
235 }
236
237 if err := os.Remove(path + journalSuffix); err != nil {
238 return fmt.Errorf("failed to remove root journal: %w", err)
239 }
240 return nil
241}
242
243// readHeaderRegion reads a CAR file's header region: the full varint-prefixed
244// bytes (region), the header bytes alone, and the varint prefix length.
245func readHeaderRegion(f *os.File) (region, header []byte, varintLen int, err error) {
246 hlen, err := binary.ReadUvarint(bufio.NewReaderSize(io.NewSectionReader(f, 0, binary.MaxVarintLen64), binary.MaxVarintLen64))
247 if err != nil {
248 return nil, nil, 0, fmt.Errorf("failed to read CAR header length: %w", err)
249 }
250 prefix := binary.AppendUvarint(nil, hlen)
251
252 header = make([]byte, hlen)
253 if _, err := f.ReadAt(header, int64(len(prefix))); err != nil {
254 return nil, nil, 0, fmt.Errorf("failed to read CAR header: %w", err)
255 }
256 return append(prefix, header...), header, len(prefix), nil
257}
258
259// recoverJournal replays a root-patch journal left behind by a crash: the
260// journal holds the complete intended header region, which is rewritten at
261// the start of the CAR file. Idempotent; a no-op if no journal exists.
262func recoverJournal(path string) error {
263 jpath := path + journalSuffix
264 region, err := os.ReadFile(jpath)
265 if err != nil {
266 if os.IsNotExist(err) {
267 return nil
268 }
269 return fmt.Errorf("failed to read root journal: %w", err)
270 }
271
272 // the journal is written atomically, so it should always parse; treat
273 // anything else as corruption we must not replay
274 if _, err := car.ReadHeader(bufio.NewReader(bytes.NewReader(region))); err != nil {
275 return fmt.Errorf("root journal is corrupt: %w", err)
276 }
277
278 f, err := os.OpenFile(path, os.O_WRONLY, 0o644)
279 if err != nil {
280 if os.IsNotExist(err) {
281 // orphan journal for a CAR that no longer exists
282 return os.Remove(jpath)
283 }
284 return fmt.Errorf("failed to open CAR for journal recovery: %w", err)
285 }
286 defer f.Close()
287
288 if _, err := f.WriteAt(region, 0); err != nil {
289 return fmt.Errorf("failed to replay root journal: %w", err)
290 }
291 if err := f.Sync(); err != nil {
292 return fmt.Errorf("failed to fsync CAR after journal recovery: %w", err)
293 }
294 if err := os.Remove(jpath); err != nil {
295 return err
296 }
297 carJournalRecoveries.Inc()
298 return nil
299}
300
301// RecoverAllJournals replays any root-patch journals left behind by a crash.
302// Called once at startup.
303func (s *CarStore) RecoverAllJournals() error {
304 journals, err := filepath.Glob(filepath.Join(s.dir, "*"+journalSuffix))
305 if err != nil {
306 return fmt.Errorf("failed to list root journals: %w", err)
307 }
308 for _, jpath := range journals {
309 carPath := strings.TrimSuffix(jpath, journalSuffix)
310 if err := recoverJournal(carPath); err != nil {
311 return fmt.Errorf("failed to recover journal for %s: %w", carPath, err)
312 }
313 }
314 return nil
315}
316
317// writeFileAtomic writes data to path via a temp file + fsync + rename.
318func writeFileAtomic(path string, data []byte) error {
319 tmp := path + ".tmp"
320 f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o644)
321 if err != nil {
322 return err
323 }
324 if _, err := f.Write(data); err != nil {
325 f.Close()
326 os.Remove(tmp)
327 return err
328 }
329 if err := f.Sync(); err != nil {
330 f.Close()
331 os.Remove(tmp)
332 return err
333 }
334 if err := f.Close(); err != nil {
335 os.Remove(tmp)
336 return err
337 }
338 if err := os.Rename(tmp, path); err != nil {
339 os.Remove(tmp)
340 return err
341 }
342 return nil
343}
344
345// DeleteCar removes the CAR file (and any root journal) for a DID.
346func (s *CarStore) DeleteCar(did string) error {
347 if err := os.Remove(s.carPath(did) + journalSuffix); err != nil && !os.IsNotExist(err) {
348 return fmt.Errorf("failed to delete root journal: %w", err)
349 }
350 err := os.Remove(s.carPath(did))
351 if err != nil && !os.IsNotExist(err) {
352 return fmt.Errorf("failed to delete CAR: %w", err)
353 }
354 return nil
355}
356
357// TrashCar moves a DID's CAR into the trash directory, where it is retained
358// until the deletion sweeper purges it. Used instead of DeleteCar when an
359// account is deleted upstream, so a mistaken or malicious deletion does not
360// instantly destroy the backup. A no-op if the repo has no CAR.
361func (s *CarStore) TrashCar(did string) error {
362 path := s.carPath(did)
363
364 // fold any pending root patch into the file before it leaves the store
365 if err := recoverJournal(path); err != nil {
366 return fmt.Errorf("failed to recover root journal before trashing: %w", err)
367 }
368
369 if err := os.Rename(path, s.trashPath(did)); err != nil {
370 if os.IsNotExist(err) {
371 return nil
372 }
373 return fmt.Errorf("failed to move CAR to trash: %w", err)
374 }
375 return nil
376}
377
378// DeleteTrash permanently removes a DID's trashed CAR.
379func (s *CarStore) DeleteTrash(did string) error {
380 err := os.Remove(s.trashPath(did))
381 if err != nil && !os.IsNotExist(err) {
382 return fmt.Errorf("failed to delete trashed CAR: %w", err)
383 }
384 carTrashPurged.Inc()
385 return nil
386}
387
388// VerifyCar re-reads a repo's CAR from disk and checks that it is intact:
389// the header parses with a single root matching expectedHead, and every
390// block's data rehashes to its stored CID (go-car does not verify hashes on
391// read). Any error means the on-disk copy can no longer be trusted and the
392// repo should be resynced.
393func (s *CarStore) VerifyCar(did string, expectedHead string) error {
394 f, err := os.Open(s.carPath(did))
395 if err != nil {
396 return fmt.Errorf("failed to open CAR: %w", err)
397 }
398 defer f.Close()
399
400 cr, err := car.NewCarReader(bufio.NewReaderSize(f, 1<<20))
401 if err != nil {
402 return fmt.Errorf("failed to read CAR header: %w", err)
403 }
404 if cr.Header.Version != 1 {
405 return fmt.Errorf("unsupported CAR version: %d", cr.Header.Version)
406 }
407 if len(cr.Header.Roots) != 1 {
408 return fmt.Errorf("expected exactly one root in CAR header, got %d", len(cr.Header.Roots))
409 }
410 if root := cr.Header.Roots[0].String(); root != expectedHead {
411 return fmt.Errorf("CAR root %s does not match expected head %s", root, expectedHead)
412 }
413
414 seen := false
415 for {
416 blk, err := cr.Next()
417 if err == io.EOF {
418 break
419 }
420 if err != nil {
421 return fmt.Errorf("failed to read block: %w", err)
422 }
423 computed, err := blk.Cid().Prefix().Sum(blk.RawData())
424 if err != nil {
425 return fmt.Errorf("failed to hash block %s: %w", blk.Cid(), err)
426 }
427 if !computed.Equals(blk.Cid()) {
428 return fmt.Errorf("block %s fails hash verification (computed %s)", blk.Cid(), computed)
429 }
430 if blk.Cid().Equals(cr.Header.Roots[0]) {
431 seen = true
432 }
433 }
434 if !seen {
435 return fmt.Errorf("root block %s not present in CAR", expectedHead)
436 }
437 return nil
438}
439
440// HasCar reports whether a CAR file exists for the DID.
441func (s *CarStore) HasCar(did string) bool {
442 _, err := os.Stat(s.carPath(did))
443 return err == nil
444}
445
446// headCidFromCar returns the first root CID from a CAR file's header
447// (for repo CARs this is the commit block CID).
448func headCidFromCar(carBytes []byte) (string, error) {
449 h, err := car.ReadHeader(bufio.NewReader(bytes.NewReader(carBytes)))
450 if err != nil {
451 return "", fmt.Errorf("failed to read CAR header: %w", err)
452 }
453 if h.Version != 1 {
454 return "", fmt.Errorf("unsupported CAR version: %d", h.Version)
455 }
456 if len(h.Roots) < 1 {
457 return "", fmt.Errorf("CAR file missing root CID")
458 }
459 return h.Roots[0].String(), nil
460}