| 1 | package dbdoor |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "net/http" |
| 10 | "net/http/httptest" |
| 11 | "os" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 16 | ) |
| 17 | |
| 18 | // addBatch inserts one more batch into src and replicates it into dir. |
| 19 | func addBatch(t *testing.T, src, dir string, batch int) { |
| 20 | t.Helper() |
| 21 | db, err := sql.Open("sqlite", src) |
| 22 | if err != nil { |
| 23 | t.Fatal(err) |
| 24 | } |
| 25 | for i := 0; i < dbdoortest.Rows; i++ { |
| 26 | if _, err := db.Exec(`INSERT INTO t (body) VALUES (?)`, fmt.Sprintf("row-%d-%d", batch, i)); err != nil { |
| 27 | db.Close() |
| 28 | t.Fatal(err) |
| 29 | } |
| 30 | } |
| 31 | db.Close() |
| 32 | dbdoortest.Sync(t, src, dir) |
| 33 | } |
| 34 | |
| 35 | // pause returns a time in a whole second of its own: after every file |
| 36 | // written so far, and before the next one. RestoreAsOf rounds its time down |
| 37 | // to the second, as S3's listing does. |
| 38 | func pause() time.Time { |
| 39 | time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(time.Second + 10*time.Millisecond))) |
| 40 | at := time.Now() |
| 41 | time.Sleep(time.Until(at.Truncate(time.Second).Add(time.Second + 10*time.Millisecond))) |
| 42 | return at |
| 43 | } |
| 44 | |
| 45 | // A copy as of T holds what was replicated before T and nothing after it. |
| 46 | func TestRestoreAsOfStopsAtTheTime(t *testing.T) { |
| 47 | dir := t.TempDir() |
| 48 | src := dbdoortest.Replicate(t, dir, 2) |
| 49 | at := pause() |
| 50 | addBatch(t, src, dir, 2) |
| 51 | c := NewClient(newTestDoor(t, dir).do) |
| 52 | latest := collect(t, c, 0, 0) |
| 53 | |
| 54 | got, err := RestoreAsOf(context.Background(), c, t.TempDir(), at) |
| 55 | if err != nil { |
| 56 | t.Fatal(err) |
| 57 | } |
| 58 | defer os.RemoveAll(got.Dir) |
| 59 | if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows { |
| 60 | t.Fatalf("copy as of %v holds %d rows, want %d (the third batch came after)", at, rows, 2*dbdoortest.Rows) |
| 61 | } |
| 62 | if got.TXID >= uint64(latest[len(latest)-1].MaxTXID) { |
| 63 | t.Fatalf("TXID = %d, want one before the newest (%d)", got.TXID, latest[len(latest)-1].MaxTXID) |
| 64 | } |
| 65 | if !got.At.Before(at) { |
| 66 | t.Fatalf("At = %v, want a time before the requested %v", got.At, at) |
| 67 | } |
| 68 | } |
| 69 | |
| 70 | // A time after the newest write is simply the newest copy. |
| 71 | func TestRestoreAsOfAfterTheNewestWrite(t *testing.T) { |
| 72 | dir := t.TempDir() |
| 73 | dbdoortest.Replicate(t, dir, 2) |
| 74 | got, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), pause()) |
| 75 | if err != nil { |
| 76 | t.Fatal(err) |
| 77 | } |
| 78 | defer os.RemoveAll(got.Dir) |
| 79 | if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows { |
| 80 | t.Fatalf("restored %d rows, want %d", rows, 2*dbdoortest.Rows) |
| 81 | } |
| 82 | } |
| 83 | |
| 84 | // A time before anything the replica keeps is its own answer, naming the |
| 85 | // oldest point it does keep, and not "nothing replicated". |
| 86 | func TestRestoreAsOfBeforeTheHistory(t *testing.T) { |
| 87 | at := time.Now().Add(-time.Hour) |
| 88 | dir := t.TempDir() |
| 89 | dbdoortest.Replicate(t, dir, 2) |
| 90 | parent := t.TempDir() |
| 91 | _, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, dir).do), parent, at) |
| 92 | if !errors.Is(err, ErrBeforeHistory) { |
| 93 | t.Fatalf("RestoreAsOf before the history = %v, want ErrBeforeHistory", err) |
| 94 | } |
| 95 | if errors.Is(err, ErrNothingReplicated) { |
| 96 | t.Fatalf("RestoreAsOf before the history = %v, must not read as nothing replicated", err) |
| 97 | } |
| 98 | var bh *BeforeHistoryError |
| 99 | if !errors.As(err, &bh) || bh.Oldest.IsZero() || !bh.Oldest.After(at) { |
| 100 | t.Fatalf("RestoreAsOf before the history = %#v, want the oldest kept point, after %v", err, at) |
| 101 | } |
| 102 | assertEmptyDir(t, parent) |
| 103 | |
| 104 | // The point it names is one a person can ask for. |
| 105 | got, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), bh.Oldest) |
| 106 | if err != nil { |
| 107 | t.Fatalf("RestoreAsOf at the named oldest point %v = %v, want a copy", bh.Oldest, err) |
| 108 | } |
| 109 | os.RemoveAll(got.Dir) |
| 110 | } |
| 111 | |
| 112 | // An empty replica is still "nothing replicated", with or without a time. |
| 113 | func TestRestoreAsOfEmptyReplica(t *testing.T) { |
| 114 | parent := t.TempDir() |
| 115 | _, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, t.TempDir()).do), parent, time.Now()) |
| 116 | if !errors.Is(err, ErrNothingReplicated) { |
| 117 | t.Fatalf("RestoreAsOf of an empty replica = %v, want ErrNothingReplicated", err) |
| 118 | } |
| 119 | assertEmptyDir(t, parent) |
| 120 | } |
| 121 | |
| 122 | // A listing with no times (the claim-mode door sends none) must refuse: |
| 123 | // litestream reads a zero time as "before T" and would hand back the |
| 124 | // newest copy labelled as the requested one. |
| 125 | func TestRestoreAsOfRefusesAListingWithoutTimes(t *testing.T) { |
| 126 | dir := t.TempDir() |
| 127 | dbdoortest.Replicate(t, dir, 2) |
| 128 | untimed := relist(newTestDoor(t, dir), func(f *LTXFile) { f.CreatedAt = time.Time{} }) |
| 129 | parent := t.TempDir() |
| 130 | _, err := RestoreAsOf(context.Background(), NewClient(untimed), parent, time.Now()) |
| 131 | if !errors.Is(err, ErrNoTimes) { |
| 132 | t.Fatalf("RestoreAsOf over an untimed listing = %v, want ErrNoTimes", err) |
| 133 | } |
| 134 | assertEmptyDir(t, parent) |
| 135 | |
| 136 | // Without a time, the same listing restores as before. |
| 137 | got, err := RestoreTo(context.Background(), NewClient(untimed), t.TempDir()) |
| 138 | if err != nil { |
| 139 | t.Fatalf("RestoreTo over an untimed listing = %v, want the newest copy", err) |
| 140 | } |
| 141 | os.RemoveAll(got.Dir) |
| 142 | } |
| 143 | |
| 144 | // relist serves door's listings with each file passed through edit. |
| 145 | func relist(door *testDoor, edit func(*LTXFile)) Doer { |
| 146 | return func(ctx context.Context, tail string, h http.Header) (*http.Response, error) { |
| 147 | res, err := door.do(ctx, tail, h) |
| 148 | if err != nil || res.StatusCode != http.StatusOK || h.Get("Range") != "" { |
| 149 | return res, err |
| 150 | } |
| 151 | var list ListResponse |
| 152 | if err := json.NewDecoder(res.Body).Decode(&list); err != nil { |
| 153 | return nil, err |
| 154 | } |
| 155 | for i := range list.Files { |
| 156 | edit(&list.Files[i]) |
| 157 | } |
| 158 | rec := httptest.NewRecorder() |
| 159 | json.NewEncoder(rec).Encode(list) |
| 160 | return rec.Result(), nil |
| 161 | } |
| 162 | } |
| 163 | |
| 164 | // S3 lists LastModified to the whole second, rounded down, so a file |
| 165 | // uploaded just after T in the same second lists as before T. The copy must |
| 166 | // still leave it out. |
| 167 | func TestRestoreAsOfWithSecondResolutionTimes(t *testing.T) { |
| 168 | dir := t.TempDir() |
| 169 | src := dbdoortest.Replicate(t, dir, 2) |
| 170 | // Start just after a whole second, so batch three lands in the same |
| 171 | // second as T but after it. |
| 172 | time.Sleep(time.Until(time.Now().Truncate(time.Second).Add(time.Second + 100*time.Millisecond))) |
| 173 | at := time.Now() |
| 174 | time.Sleep(50 * time.Millisecond) |
| 175 | addBatch(t, src, dir, 2) |
| 176 | if done := time.Now(); done.Truncate(time.Second) != at.Truncate(time.Second) { |
| 177 | t.Skipf("batch three took until %v, past T's second; the case needs it inside", done) |
| 178 | } |
| 179 | c := NewClient(relist(newTestDoor(t, dir), func(f *LTXFile) { f.CreatedAt = f.CreatedAt.Truncate(time.Second) })) |
| 180 | got, err := RestoreAsOf(context.Background(), c, t.TempDir(), at) |
| 181 | if err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | defer os.RemoveAll(got.Dir) |
| 185 | if rows := countRows(t, got.Path); rows != 2*dbdoortest.Rows { |
| 186 | t.Fatalf("copy as of %v holds %d rows, want %d: a write after T is in it", at, rows, 2*dbdoortest.Rows) |
| 187 | } |
| 188 | } |
| 189 | |
| 190 | // Go's zero time means "the newest copy" to RestoreAsOf and to litestream. |
| 191 | // A requested time at or near it (year 1, or a fraction of a second into |
| 192 | // it that rounds down to zero) must still be a time before the history, |
| 193 | // never the newest copy. Found in Astra's review. The zero instant itself, |
| 194 | // in any zone, is "the newest copy" by contract; parseDBAt keeps a typed |
| 195 | // one from reaching here. |
| 196 | func TestRestoreAsOfNeverTreatsAnEarlyTimeAsTheNewest(t *testing.T) { |
| 197 | dir := t.TempDir() |
| 198 | dbdoortest.Replicate(t, dir, 2) |
| 199 | for _, at := range []time.Time{ |
| 200 | time.Time{}.Add(500 * time.Millisecond), |
| 201 | time.Time{}.Add(time.Hour), |
| 202 | } { |
| 203 | _, err := RestoreAsOf(context.Background(), NewClient(newTestDoor(t, dir).do), t.TempDir(), at) |
| 204 | if !errors.Is(err, ErrBeforeHistory) { |
| 205 | t.Errorf("RestoreAsOf(%v) = %v, want ErrBeforeHistory", at, err) |
| 206 | } |
| 207 | } |
| 208 | } |
| 209 | |
amadan