carlos / platform Public

Clone
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package dbdoor
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "net/http"
10 "os"
11 "path/filepath"
12 "sync"
13 "time"
14
15 "github.com/benbjohnson/litestream"
16 "github.com/superfly/ltx"
17
18 "github.com/carlosframework/platform/internal/lease"
19)
20
21var (
22 // ErrNothingReplicated: the replica holds no restorable state yet.
23 // litestream says so two ways (ErrNoSnapshots, ErrTxNotAvailable), the
24 // pair lease.Restore already treats as "nothing to restore".
25 ErrNothingReplicated = errors.New("dbdoor: nothing has been replicated yet")
26 // ErrCheckFailed: the copy restored, but lease.CheckDB refused it. It
27 // is never handed over.
28 ErrCheckFailed = errors.New("dbdoor: the restored copy failed its check")
29 // ErrHistoryGone: a restore target older than the replica's retained
30 // history (seeded-instances spec §4 step 4).
31 ErrHistoryGone = errors.New("dbdoor: history before the target is no longer kept")
32 // ErrBeforeHistory: an as-of restore asked for a time earlier than
33 // anything the replica keeps. RestoreAsOf returns it as a
34 // *BeforeHistoryError naming the oldest point there is.
35 ErrBeforeHistory = errors.New("dbdoor: the time is earlier than the history the replica keeps")
36 // ErrNoTimes: an as-of restore met a listing with no file times (the
37 // claim-mode door sends none). litestream reads a zero time as "before
38 // T" and would restore the newest copy, so this refuses instead.
39 ErrNoTimes = errors.New("dbdoor: the replica's listing carries no times, so it cannot be read as of a time")
40)
41
42// EarliestAsOf is the earliest time RestoreAsOf asks litestream for. Any
43// earlier request is moved to it: it is before every replica's history,
44// and unlike Go's zero time it cannot mean "the newest copy".
45var EarliestAsOf = time.Unix(0, 0).UTC()
46
47// BeforeHistoryError is ErrBeforeHistory with the oldest point the replica
48// keeps: the earliest time an as-of restore can ask for, the first whole
49// second after the oldest file a restore could start from was listed.
50type BeforeHistoryError struct {
51 Oldest time.Time
52}
53
54func (e *BeforeHistoryError) Error() string {
55 return fmt.Sprintf("%v (the oldest point kept is %s)", ErrBeforeHistory, e.Oldest.UTC().Format(time.RFC3339))
56}
57
58func (e *BeforeHistoryError) Is(target error) bool { return target == ErrBeforeHistory }
59
60// Restored is a private copy of a replica on this machine.
61type Restored struct {
62 // Path is the restored database. It lives inside Dir.
63 Path string
64 // Dir is the private 0700 directory holding Path and any -wal/-shm a
65 // read-only open makes. The caller removes it.
66 Dir string
67 // TXID is the newest replicated transaction in the copy, and At is
68 // that transaction's time from its LTX header: the time of the write,
69 // not of a later compaction (spec §8.1).
70 TXID uint64
71 At time.Time
72}
73
74// restoredName is the copy's file name inside its private directory.
75const restoredName = "replica.db"
76
77// checkDB is lease.CheckDBContext, the one check a disaster recovery leans on. A
78// variable only so a test can make it refuse.
79var checkDB = lease.CheckDBContext
80
81// RestoreTo restores the replica c reads into a new 0700 directory under
82// parent (os.TempDir() when empty) and proves it with lease.CheckDB.
83//
84// The three conditions spec §12 rests on hold here: c is not a
85// ReplicaClientV3, RestoreOptions.Follow is never set, and CheckDB runs on
86// the result rather than relying on litestream's own IntegrityCheck.
87//
88// A planned file can vanish under the restore when the writer's L0
89// retention fires, and litestream's ResumableReader treats that as fatal.
90// So the retry wraps the WHOLE Restore, once: a fresh plan is the only
91// recovery. On any failure the directory is removed, so no copy of
92// production data is left behind.
93func RestoreTo(ctx context.Context, c *Client, parent string) (*Restored, error) {
94 return restoreAt(ctx, c, parent, 0, time.Time{})
95}
96
97// RestoreToTarget is RestoreTo at a target transaction: 0 is the latest.
98//
99// A nonzero target restores exactly to target or not at all. litestream
100// v0.5.14's CalcRestorePlan skips every file whose MaxTXID is past target
101// and refuses unless the chain it builds ends exactly at target, so the
102// copy never reaches past it and Restored.TXID is target. A target no
103// retained chain ends at (older than the retained history, inside a
104// compacted file, or past the latest transaction) is refused as
105// ErrHistoryGone. Ask for a PlanRestore's TXID: that restore produces
106// exactly the plan's RestoredBytes.
107func RestoreToTarget(ctx context.Context, c *Client, parent string, target uint64) (*Restored, error) {
108 return restoreAt(ctx, c, parent, target, time.Time{})
109}
110
111// RestoreAsOf is RestoreTo stopped at a point in time: the copy holds every
112// file replicated before at and nothing after it. A zero at is the newest
113// copy, which is RestoreTo.
114//
115// The plan is litestream's own (Timestamp set, TXID not). It keeps a file
116// only if its listed time is before at, and the door lists S3's
117// LastModified, the upload time, which is never earlier than the writes a
118// file holds. at is rounded down to its second first (restoreOnce says
119// why). So the copy can be older than at, never newer. Restored.At
120// says what it actually holds.
121func RestoreAsOf(ctx context.Context, c *Client, parent string, at time.Time) (*Restored, error) {
122 // A requested time that rounds down to Go's zero time would read as
123 // "the newest copy" to litestream. Anything that early is before every
124 // replica's history, so it is asked as the Unix epoch, which still is.
125 if !at.IsZero() && at.Truncate(time.Second).Before(EarliestAsOf) {
126 at = EarliestAsOf
127 }
128 return restoreAt(ctx, c, parent, 0, at)
129}
130
131// restoreAt is the one restore behind RestoreTo, RestoreToTarget and
132// RestoreAsOf. At most one of target and at is set: a target transaction
133// (RestoreToTarget) or an as-of time (RestoreAsOf). Neither is the newest
134// copy. Each keeps its own refusal: a target no chain ends at is
135// ErrHistoryGone, and a time before the kept history is ErrBeforeHistory.
136func restoreAt(ctx context.Context, c *Client, parent string, target uint64, at time.Time) (*Restored, error) {
137 dir, err := os.MkdirTemp(parent, "carlos-db-")
138 if err != nil {
139 return nil, err
140 }
141 // MkdirTemp already creates 0700; say it anyway, because the privacy of
142 // everything restored rests on this one mode.
143 if err := os.Chmod(dir, 0o700); err != nil {
144 os.RemoveAll(dir)
145 return nil, err
146 }
147 out := filepath.Join(dir, restoredName)
148 var rec *recorder
149 for attempt := 0; ; attempt++ {
150 rec = &recorder{Client: c, dir: dir, at: at}
151 err = restoreOnce(ctx, rec, out, target, at)
152 if err == nil {
153 break
154 }
155 removeCopy(out)
156 rec.mu.Lock()
157 checkFailed := rec.checkFailed
158 rec.mu.Unlock()
159 if checkFailed {
160 os.RemoveAll(dir)
161 if cerr := ctx.Err(); cerr != nil {
162 return nil, cerr
163 }
164 return nil, ErrCheckFailed
165 }
166 vanished := errors.Is(err, os.ErrNotExist) || rec.sawVanish()
167 if attempt == 0 && vanished && ctx.Err() == nil {
168 continue
169 }
170 os.RemoveAll(dir)
171 // A stop reaches litestream's local half as a broken read, which
172 // its compactor reports without wrapping. Name the real cause.
173 if cerr := ctx.Err(); cerr != nil {
174 return nil, cerr
175 }
176 switch {
177 case errors.Is(err, ErrNoTimes):
178 return nil, ErrNoTimes
179 case errors.Is(err, litestream.ErrNoSnapshots) || errors.Is(err, litestream.ErrTxNotAvailable):
180 if target != 0 {
181 return nil, ErrHistoryGone
182 }
183 if oldest := rec.oldestStart(); !at.IsZero() && !oldest.IsZero() {
184 if !oldest.Before(at.Truncate(time.Second)) {
185 // The first whole second after the oldest file: the
186 // earliest --at that, rounded down, still includes it.
187 return nil, &BeforeHistoryError{Oldest: oldest.Truncate(time.Second).Add(time.Second)}
188 }
189 return nil, err
190 }
191 return nil, ErrNothingReplicated
192 case vanished && !errors.Is(err, os.ErrNotExist):
193 return nil, fmt.Errorf("%w: %w", os.ErrNotExist, err)
194 }
195 return nil, err
196 }
197 if err := ctx.Err(); err != nil {
198 os.RemoveAll(dir)
199 return nil, err
200 }
201 if err := checkDB(ctx, out); err != nil {
202 os.RemoveAll(dir)
203 if cerr := ctx.Err(); cerr != nil {
204 return nil, cerr
205 }
206 return nil, fmt.Errorf("%w: %v", ErrCheckFailed, err)
207 }
208 // A stop that lands as the check finishes still withdraws the copy.
209 if err := ctx.Err(); err != nil {
210 os.RemoveAll(dir)
211 return nil, err
212 }
213 txid, at := rec.newest()
214 return &Restored{Path: out, Dir: dir, TXID: txid, At: at}, nil
215}
216
217func restoreOnce(ctx context.Context, client litestream.ReplicaClient, out string, target uint64, at time.Time) error {
218 opt := litestream.NewRestoreOptions()
219 opt.OutputPath = out
220 // A target transaction (RestoreToTarget); zero is none.
221 opt.TXID = ltx.TXID(target)
222 // S3 lists LastModified to the whole second, rounded down: a file
223 // uploaded at 14:00:00.7 lists as 14:00:00, before a T of 14:00:00.5,
224 // though it holds writes after T. Rounding T down to its second keeps
225 // it out: a listed time before a whole second means an upload before it.
226 // A zero at stays zero: no time bound.
227 opt.Timestamp = at.Truncate(time.Second)
228 // Follow stays false (spec §12 condition ii); IntegrityCheck is left at
229 // litestream's default and CheckDB runs after (condition iii).
230 return litestream.NewReplicaWithClient(nil, client).Restore(ctx, opt)
231}
232
233func removeCopy(out string) {
234 for _, p := range []string{out, out + ".tmp", out + "-wal", out + "-shm"} {
235 _ = os.Remove(p)
236 }
237}
238
239// recorder sees every file a Restore opens, for two reasons: a file that
240// answered os.ErrNotExist is what makes the restore worth a second plan,
241// whatever litestream wrapped it in on the way out; and the header of the
242// newest file opened is the copy's "as of" (R1).
243type recorder struct {
244 *Client
245 dir string // the private directory each file is downloaded into
246 mu sync.Mutex
247 vanished bool
248 checkFailed bool
249 at time.Time // the as-of time; zero for the newest copy
250 oldest time.Time // earliest time of a file a restore can start from
251 maxTXID ltx.TXID
252 header *bytes.Buffer // first ltx.HeaderSize bytes of the maxTXID file
253}
254
255// LTXFiles passes the listing through, and for an as-of restore checks it
256// first: every file must carry a time, and the earliest file a restore can
257// start from (any snapshot, or a file beginning at the first transaction)
258// is remembered so a time before it can be named.
259func (r *recorder) LTXFiles(ctx context.Context, level int, seek ltx.TXID, useMetadata bool) (ltx.FileIterator, error) {
260 itr, err := r.Client.LTXFiles(ctx, level, seek, useMetadata)
261 if err != nil || r.at.IsZero() {
262 return itr, err
263 }
264 var infos []*ltx.FileInfo
265 for itr.Next() {
266 infos = append(infos, itr.Item())
267 }
268 err = itr.Err()
269 if cerr := itr.Close(); err == nil {
270 err = cerr
271 }
272 if err != nil {
273 return nil, err
274 }
275 r.mu.Lock()
276 defer r.mu.Unlock()
277 for _, f := range infos {
278 if f.CreatedAt.IsZero() {
279 return nil, ErrNoTimes
280 }
281 if level == litestream.SnapshotLevel || f.MinTXID == 1 {
282 if r.oldest.IsZero() || f.CreatedAt.Before(r.oldest) {
283 r.oldest = f.CreatedAt
284 }
285 }
286 }
287 return ltx.NewFileInfoSliceIterator(infos), nil
288}
289
290func (r *recorder) oldestStart() time.Time {
291 r.mu.Lock()
292 defer r.mu.Unlock()
293 return r.oldest
294}
295
296func (r *recorder) OpenLTXFile(ctx context.Context, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
297 // The first open goes through download's loop too, so a full door's
298 // 429 waits out its Retry-After here rather than failing back to
299 // litestream, whose own quick retries ignore it.
300 rc, err := r.download(ctx, nil, level, minTXID, maxTXID, offset, size)
301 if err != nil {
302 return nil, err
303 }
304 if offset != 0 {
305 return rc, nil
306 }
307 r.mu.Lock()
308 defer r.mu.Unlock()
309 if maxTXID < r.maxTXID {
310 return rc, nil
311 }
312 r.maxTXID = maxTXID
313 r.header = &bytes.Buffer{}
314 return &headerTee{ReadCloser: rc, buf: r.header}, nil
315}
316
317// dbDownloadMaxStalls is how many reopens in a row may deliver no byte
318// before a download gives up. Progress resets the count, so a slow but
319// moving link always finishes.
320const dbDownloadMaxStalls = 5
321
322// download reads one file whole into the private directory and closes the
323// stream before the restore reads a byte of it. litestream's compactor
324// holds every planned file open until the restore ends, so streaming
325// straight through would hold one door stream per planned file, and the
326// door bounds a person's concurrent streams (apidb.go). Downloaded, a
327// restore holds one stream at a time.
328//
329// A connection that drops part way is resumed here, from the bytes already
330// kept, rather than handed to litestream: its resumable reader would start
331// the file again from the requested offset, and a large file over a flaky
332// link would never finish. The file goes when the restore closes it; on
333// failure, with the whole directory.
334func (r *recorder) download(ctx context.Context, rc io.ReadCloser, level int, minTXID, maxTXID ltx.TXID, offset, size int64) (io.ReadCloser, error) {
335 f, err := os.CreateTemp(r.dir, "ltx-*")
336 if err != nil {
337 if rc != nil {
338 rc.Close()
339 }
340 return nil, err
341 }
342 fail := func(err error) (io.ReadCloser, error) {
343 f.Close()
344 os.Remove(f.Name())
345 return nil, err
346 }
347 var written int64
348 stalls, wait := 0, dbDownloadBackoff
349 for {
350 if rc != nil {
351 n, cerr := io.Copy(f, rc)
352 rc.Close()
353 rc = nil
354 written += n
355 if cerr == nil {
356 break
357 }
358 if n > 0 {
359 stalls, wait = 0, dbDownloadBackoff
360 } else if stalls++; stalls >= dbDownloadMaxStalls {
361 return fail(cerr)
362 }
363 }
364 if ctx.Err() != nil {
365 return fail(ctx.Err())
366 }
367 var rest int64 // 0: to the end, as RangeHeader reads it
368 if size > 0 {
369 if rest = size - written; rest <= 0 {
370 break
371 }
372 }
373 next, err := r.Client.OpenLTXFile(ctx, level, minTXID, maxTXID, offset+written, rest)
374 if err == nil {
375 rc = next
376 continue
377 }
378 switch {
379 case errors.Is(err, os.ErrNotExist):
380 r.mu.Lock()
381 r.vanished = true
382 r.mu.Unlock()
383 return fail(err)
384 case ctx.Err() != nil:
385 return fail(ctx.Err())
386 case !transientOpen(err):
387 return fail(err)
388 }
389 // A busy door or a dropped connection: wait and ask again for the
390 // same remaining range, keeping every byte already here. It counts
391 // as an attempt that delivered nothing.
392 if stalls++; stalls >= dbDownloadMaxStalls {
393 return fail(err)
394 }
395 d := wait
396 var door *DoorError
397 if errors.As(err, &door) && door.RetryAfter > 0 {
398 d = door.RetryAfter
399 }
400 wait = min(wait*2, dbDownloadBackoffMax)
401 select {
402 case <-ctx.Done():
403 return fail(ctx.Err())
404 case <-time.After(d):
405 }
406 }
407 if _, err := f.Seek(0, io.SeekStart); err != nil {
408 return fail(err)
409 }
410 if err := r.verifyPinnedSnapshot(ctx, f, level, minTXID, maxTXID, offset, written); err != nil {
411 r.mu.Lock()
412 r.checkFailed = true
413 r.mu.Unlock()
414 return fail(err)
415 }
416 return &downloaded{ctx: ctx, File: f}, nil
417}
418
419func (r *recorder) verifyPinnedSnapshot(ctx context.Context, f *os.File, level int, lo, hi ltx.TXID, offset, written int64) error {
420 r.handles.mu.Lock()
421 meta := r.handles.files[GetTail(level, uint64(lo), uint64(hi))]
422 r.handles.mu.Unlock()
423 if meta.SnapshotCRC64 == nil {
424 return nil
425 }
426 if offset != 0 || written != meta.Size {
427 return ErrCheckFailed
428 }
429 dec := ltx.NewDecoder(&downloaded{ctx: ctx, File: f})
430 if err := dec.Verify(); err != nil || uint64(dec.Trailer().FileChecksum) != *meta.SnapshotCRC64 {
431 return ErrCheckFailed
432 }
433 _, err := f.Seek(0, io.SeekStart)
434 return err
435}
436
437// downloaded is a file read once and removed on Close.
438type downloaded struct {
439 ctx context.Context
440 *os.File
441}
442
443// Read checks the restore's context first. Once every file is downloaded,
444// litestream's compaction reads only these files and checks no context of
445// its own, so this is where Ctrl-C reaches it.
446func (d *downloaded) Read(p []byte) (int, error) {
447 if err := d.ctx.Err(); err != nil {
448 return 0, err
449 }
450 return d.File.Read(p)
451}
452
453func (d *downloaded) Close() error {
454 err := d.File.Close()
455 if rerr := os.Remove(d.File.Name()); err == nil {
456 err = rerr
457 }
458 return err
459}
460
461func (r *recorder) sawVanish() bool {
462 r.mu.Lock()
463 defer r.mu.Unlock()
464 return r.vanished
465}
466
467// newest returns the copy's newest transaction and the time its LTX header
468// records. A header that cannot be read leaves At zero rather than guessed.
469func (r *recorder) newest() (uint64, time.Time) {
470 r.mu.Lock()
471 defer r.mu.Unlock()
472 var h ltx.Header
473 if r.header == nil || r.header.Len() < ltx.HeaderSize || h.UnmarshalBinary(r.header.Bytes()[:ltx.HeaderSize]) != nil {
474 return uint64(r.maxTXID), time.Time{}
475 }
476 return uint64(r.maxTXID), time.UnixMilli(h.Timestamp).UTC()
477}
478
479// headerTee copies the first ltx.HeaderSize bytes read through it.
480type headerTee struct {
481 io.ReadCloser
482 buf *bytes.Buffer
483}
484
485func (t *headerTee) Read(p []byte) (int, error) {
486 n, err := t.ReadCloser.Read(p)
487 if need := ltx.HeaderSize - t.buf.Len(); need > 0 && n > 0 {
488 t.buf.Write(p[:min(n, need)])
489 }
490 return n, err
491}
492
493// dbDownloadBackoff is the first wait before retrying a reopen that met a
494// busy door or a dropped connection; each further wait doubles, up to
495// dbDownloadBackoffMax. A 429's Retry-After overrides it.
496var dbDownloadBackoff = 250 * time.Millisecond
497
498const dbDownloadBackoffMax = 8 * time.Second
499
500// transientOpen reports a reopen failure worth waiting out: the door busy
501// or briefly unavailable (429, 502, 503, 504), or no answer at all (a
502// transport error). Any other refusal from the door is final.
503func transientOpen(err error) bool {
504 var door *DoorError
505 if errors.As(err, &door) {
506 switch door.Status {
507 case http.StatusTooManyRequests, http.StatusBadGateway, http.StatusServiceUnavailable, http.StatusGatewayTimeout:
508 return true
509 }
510 return false
511 }
512 return true
513}
514