Skip to content

Commit b538e42

Browse files
pedrocarlopedrocarlo
committed
use tursodb to prepare database imports
1 parent 231f7a6 commit b538e42

5 files changed

Lines changed: 97 additions & 103 deletions

File tree

‎internal/cmd/group_flag.go‎

Lines changed: 74 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -361,6 +361,42 @@ func checkpointWALBeforeUpload(file string) error {
361361
return nil
362362
}
363363

364+
// prepareTursoDBFile asks the TursoDB engine to convert the database to MVCC,
365+
// checkpoint any logical-log entries into the main file, and validate the
366+
// resulting database. Uploading only the main file is safe once all data-bearing
367+
// sidecars are empty.
368+
func prepareTursoDBFile(file string) error {
369+
output, err := exec.Command("tursodb", "-q", "-m", "list", file,
370+
"PRAGMA journal_mode = mvcc; PRAGMA wal_checkpoint(TRUNCATE); PRAGMA quick_check;").CombinedOutput()
371+
if err != nil {
372+
return fmt.Errorf("could not prepare %s for TursoDB import: %w: %s", file, err, strings.TrimSpace(string(output)))
373+
}
374+
375+
lines := strings.Fields(string(output))
376+
if len(lines) == 0 || lines[len(lines)-1] != "ok" {
377+
return fmt.Errorf("TursoDB integrity check failed for %s: %s", file, strings.TrimSpace(string(output)))
378+
}
379+
380+
format, err := sniffSQLiteFileFormat(file)
381+
if err != nil {
382+
return err
383+
}
384+
if format != fileFormatMVCC {
385+
return fmt.Errorf("TursoDB did not convert %s to MVCC format", file)
386+
}
387+
388+
for _, sidecar := range []struct{ path, hint string }{
389+
{file + "-wal", "close all connections to the database and retry the import"},
390+
{file + "-journal", "the database has a leftover rollback journal; close it cleanly and retry the import"},
391+
{tursodbLogPath(file), "close all TursoDB connections and retry the import"},
392+
} {
393+
if err := checkSidecarEmpty(sidecar.path, sidecar.hint); err != nil {
394+
return err
395+
}
396+
}
397+
return nil
398+
}
399+
364400
func checkSidecarEmpty(sidecarPath, hint string) error {
365401
info, err := os.Stat(sidecarPath)
366402
if errors.Is(err, os.ErrNotExist) {
@@ -389,6 +425,9 @@ func handleDBFileAWS(file string, cipher string) (*turso.DBSeed, error) {
389425
}
390426

391427
if format == fileFormatRollback {
428+
if err := checkSQLiteAvailable(); err != nil {
429+
return nil, err
430+
}
392431
// The server only accepts WAL or MVCC format files. Converting to WAL
393432
// is exactly the remediation the error message used to instruct users
394433
// to run themselves, and sqlite3 is already a hard requirement here.
@@ -403,6 +442,9 @@ func handleDBFileAWS(file string, cipher string) (*turso.DBSeed, error) {
403442

404443
switch format {
405444
case fileFormatWAL:
445+
if err := checkSQLiteAvailable(); err != nil {
446+
return nil, err
447+
}
406448
if err := checkpointWALBeforeUpload(file); err != nil {
407449
return nil, err
408450
}
@@ -411,27 +453,13 @@ func handleDBFileAWS(file string, cipher string) (*turso.DBSeed, error) {
411453
}
412454
case fileFormatMVCC:
413455
// sqlite3-based checks can't run on MVCC (tursodb format) files: the
414-
// sqlite3 binary reports them as not-a-database. The server verifies
415-
// the file with the tursodb engine after upload.
456+
// sqlite3 binary reports them as not-a-database.
416457
if !tursoDBFlag {
417458
return nil, fmt.Errorf("%s is in tursodb (MVCC) format and can only be imported into a tursodb database", file)
418459
}
419460
if cipher != "" {
420461
return nil, errors.New("remote encryption is not supported when importing tursodb (MVCC) format files")
421462
}
422-
// The upload ships only the main database file: a non-empty tursodb
423-
// logical log next to it means the file is not a fully checkpointed
424-
// snapshot and importing it would lose the log's data.
425-
if err := checkSidecarEmpty(tursodbLogPath(file), "checkpoint the database with tursodb before importing"); err != nil {
426-
return nil, err
427-
}
428-
fileInfo, err := os.Stat(file)
429-
if err != nil {
430-
return nil, fmt.Errorf("failed to get file info: %w", err)
431-
}
432-
if fileInfo.Size() > MaxAWSDBSizeBytes {
433-
return nil, errors.New("database file size exceeds maximum allowed size of 20 GB")
434-
}
435463
case fileFormatNotSQLite:
436464
isDump, err := checkIfDump(file)
437465
if err != nil {
@@ -445,6 +473,26 @@ func handleDBFileAWS(file string, cipher string) (*turso.DBSeed, error) {
445473
return nil, fmt.Errorf("file %s has an unsupported SQLite file format", file)
446474
}
447475

476+
if tursoDBFlag {
477+
if err := checkTursoDBAvailable(); err != nil {
478+
return nil, err
479+
}
480+
if format != fileFormatMVCC {
481+
fmt.Printf("Converting %s to TursoDB (MVCC) format for import.\n", file)
482+
}
483+
if err := prepareTursoDBFile(file); err != nil {
484+
return nil, err
485+
}
486+
}
487+
488+
fileInfo, err := os.Stat(file)
489+
if err != nil {
490+
return nil, fmt.Errorf("failed to get file info: %w", err)
491+
}
492+
if fileInfo.Size() > MaxAWSDBSizeBytes {
493+
return nil, errors.New("database file size exceeds maximum allowed size of 20 GB")
494+
}
495+
448496
seed := &turso.DBSeed{
449497
Type: "database_upload",
450498
Filepath: file,
@@ -457,9 +505,6 @@ func handleDBFile(client *turso.Client, file string, isAWS bool, cipher string)
457505
if err := checkFileExists(file); err != nil {
458506
return nil, err
459507
}
460-
if err := checkSQLiteAvailable(); err != nil {
461-
return nil, err
462-
}
463508

464509
if isAWS {
465510
return handleDBFileAWS(file, cipher)
@@ -474,6 +519,9 @@ func handleDBFile(client *turso.Client, file string, isAWS bool, cipher string)
474519
// produce from an MVCC (tursodb format) file
475520
return nil, fmt.Errorf("%s is in tursodb (MVCC) format and can only be imported into AWS groups", file)
476521
}
522+
if err := checkSQLiteAvailable(); err != nil {
523+
return nil, err
524+
}
477525

478526
if err := checkSQLiteFile(file); err != nil {
479527
return nil, err
@@ -507,6 +555,14 @@ func checkSQLiteAvailable() error {
507555
return err
508556
}
509557

558+
func checkTursoDBAvailable() error {
559+
_, err := exec.LookPath("tursodb")
560+
if errors.Is(err, exec.ErrNotFound) {
561+
return errors.New("could not find tursodb on your system. Please install it to import into a TursoDB database")
562+
}
563+
return err
564+
}
565+
510566
func checkSQLiteFile(file string) error {
511567
output, err := exec.Command("sqlite3", "-list", file, "pragma quick_check;").CombinedOutput()
512568

‎internal/cmd/group_flag_test.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -120,3 +120,22 @@ func TestCheckpointWALBeforeUpload(t *testing.T) {
120120
}
121121
}
122122
}
123+
124+
func TestPrepareTursoDBFile(t *testing.T) {
125+
if _, err := exec.LookPath("tursodb"); err != nil {
126+
t.Skip("tursodb not available, skipping test")
127+
}
128+
129+
dbPath := createTestDatabase(t, 10*1024)
130+
require.NoError(t, checkpointWALBeforeUpload(dbPath))
131+
require.NoError(t, prepareTursoDBFile(dbPath))
132+
133+
format, err := sniffSQLiteFileFormat(dbPath)
134+
require.NoError(t, err)
135+
require.Equal(t, fileFormatMVCC, format)
136+
for _, sidecar := range []string{dbPath + "-wal", dbPath + "-journal", tursodbLogPath(dbPath)} {
137+
if info, err := os.Stat(sidecar); err == nil {
138+
require.Zero(t, info.Size(), "%s must be empty after checkpoint", sidecar)
139+
}
140+
}
141+
}

‎internal/turso/databases.go‎

Lines changed: 3 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -242,7 +242,7 @@ func (d *DatabasesClient) Create(name, location, image, extensions, group string
242242
}
243243

244244
if isTursoServerUpload {
245-
if _, err = d.UploadDatabaseAWS(data, group, uploadFilepath, remoteEncryptionCipher, remoteEncryptionKey, useTursoDB, spinner); err != nil {
245+
if _, err = d.UploadDatabaseAWS(data, group, uploadFilepath, remoteEncryptionCipher, remoteEncryptionKey, spinner); err != nil {
246246
// Clean up the database if the upload fails
247247
if deleteErr := d.Delete(data.Database.Name); deleteErr != nil {
248248
fmt.Printf("%v", deleteErr)
@@ -263,7 +263,7 @@ func (d *DatabasesClient) Create(name, location, image, extensions, group string
263263
// This call happens in DatabasesClient.Create() above, after which it calls this function.
264264
// 2. This function creates a DB token for the newly-created DB, and then calls turso-server to upload the database file.
265265
// turso-server will perform validations on the file and 'activate' the db if everything is ok.
266-
func (d *DatabasesClient) UploadDatabaseAWS(resp *CreateDatabaseResponse, group, uploadFilepath, remoteEncryptionCipher, remoteEncryptionKey string, useTursoDB bool, spinner *prompt.SpinnerT) (*CreateDatabaseResponse, error) {
266+
func (d *DatabasesClient) UploadDatabaseAWS(resp *CreateDatabaseResponse, group, uploadFilepath, remoteEncryptionCipher, remoteEncryptionKey string, spinner *prompt.SpinnerT) (*CreateDatabaseResponse, error) {
267267
dbName := resp.Database.Name
268268
tokenTTL := 5 * time.Minute
269269
tokenProvider := func() (string, error) {
@@ -282,13 +282,7 @@ func (d *DatabasesClient) UploadDatabaseAWS(resp *CreateDatabaseResponse, group,
282282
// Upload the database file
283283
spinner.Text(fmt.Sprintf("Uploading database %s in group %s, this may take a while...", internal.Emph(resp.Database.Name), internal.Emph(group)))
284284

285-
// TursoDB databases only accept MVCC-format uploads: the WAL file's
286-
// format version bytes are rewritten to MVCC in the upload stream.
287-
upload := tursoServerClient.UploadFileMultipart
288-
if useTursoDB {
289-
upload = tursoServerClient.UploadFileMultipartMVCC
290-
}
291-
err = upload(uploadFilepath, remoteEncryptionCipher, remoteEncryptionKey, func(progressPct int, uploadedBytes int64, totalBytes int64, elapsedTime time.Duration, done bool) {
285+
err = tursoServerClient.UploadFileMultipart(uploadFilepath, remoteEncryptionCipher, remoteEncryptionKey, func(progressPct int, uploadedBytes int64, totalBytes int64, elapsedTime time.Duration, done bool) {
292286
totalSeconds := int(elapsedTime.Seconds())
293287
minutes := totalSeconds / 60
294288
seconds := totalSeconds % 60

‎internal/turso/tursoServer.go‎

Lines changed: 1 addition & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -271,19 +271,6 @@ func (i *TursoServerClient) uploadChunkWithRetry(ctx *chunkUploadContext, maxRet
271271

272272
// UploadFileMultipart uploads a database file using the multipart upload flow.
273273
func (i *TursoServerClient) UploadFileMultipart(filepath string, remoteEncryptionCipher, remoteEncryptionKey string, onUploadProgress func(progressPct int, uploadedBytes int64, totalBytes int64, elapsedTime time.Duration, done bool)) error {
274-
return i.uploadFileMultipart(filepath, false, remoteEncryptionCipher, remoteEncryptionKey, onUploadProgress)
275-
}
276-
277-
// UploadFileMultipartMVCC uploads a database file like UploadFileMultipart,
278-
// but rewrites the SQLite read/write format version bytes (header offsets
279-
// 18/19) to 255 (MVCC) in the upload stream. TursoDB databases only accept
280-
// MVCC-format uploads, and a checkpointed WAL file differs from tursodb
281-
// format in exactly those two bytes; the file on disk is left untouched.
282-
func (i *TursoServerClient) UploadFileMultipartMVCC(filepath string, remoteEncryptionCipher, remoteEncryptionKey string, onUploadProgress func(progressPct int, uploadedBytes int64, totalBytes int64, elapsedTime time.Duration, done bool)) error {
283-
return i.uploadFileMultipart(filepath, true, remoteEncryptionCipher, remoteEncryptionKey, onUploadProgress)
284-
}
285-
286-
func (i *TursoServerClient) uploadFileMultipart(filepath string, convertToMVCC bool, remoteEncryptionCipher, remoteEncryptionKey string, onUploadProgress func(progressPct int, uploadedBytes int64, totalBytes int64, elapsedTime time.Duration, done bool)) error {
287274
file, err := os.Open(filepath)
288275
if err != nil {
289276
return fmt.Errorf("failed to open file %s: %w", filepath, err)
@@ -308,12 +295,7 @@ func (i *TursoServerClient) uploadFileMultipart(filepath string, convertToMVCC b
308295
return err
309296
}
310297

311-
var reader io.ReadSeeker = file
312-
if convertToMVCC {
313-
reader = &mvccFormatReader{file: file}
314-
}
315-
316-
uploadedBytes, err := i.uploadChunks(uploadStart.UploadID, uploadStart.ChunkSize, reader, totalSize, startTime, remoteEncryptionCipher, remoteEncryptionKey, onUploadProgress)
298+
uploadedBytes, err := i.uploadChunks(uploadStart.UploadID, uploadStart.ChunkSize, file, totalSize, startTime, remoteEncryptionCipher, remoteEncryptionKey, onUploadProgress)
317299
if err != nil {
318300
return err
319301
}
@@ -373,31 +355,6 @@ func (i *TursoServerClient) startMultipartUpload(dbSize int64) (multipartUploadS
373355
return multipartUploadStart(uploadResp), nil
374356
}
375357

376-
// mvccFormatReader wraps a database file and rewrites the SQLite read/write
377-
// format version bytes (offsets 18/19) to 255 (MVCC) as the data streams
378-
// through. Seeks pass through, so chunk retries re-read patched data.
379-
type mvccFormatReader struct {
380-
file *os.File
381-
pos int64
382-
}
383-
384-
func (r *mvccFormatReader) Seek(offset int64, whence int) (int64, error) {
385-
pos, err := r.file.Seek(offset, whence)
386-
r.pos = pos
387-
return pos, err
388-
}
389-
390-
func (r *mvccFormatReader) Read(p []byte) (int, error) {
391-
n, err := r.file.Read(p)
392-
for _, formatByteOffset := range []int64{18, 19} {
393-
if formatByteOffset >= r.pos && formatByteOffset < r.pos+int64(n) {
394-
p[formatByteOffset-r.pos] = 255
395-
}
396-
}
397-
r.pos += int64(n)
398-
return n, err
399-
}
400-
401358
func (i *TursoServerClient) uploadChunks(uploadID string, chunkSize int64, file io.ReadSeeker, totalSize int64, startTime time.Time, remoteEncryptionCipher, remoteEncryptionKey string, onUploadProgress func(progressPct int, uploadedBytes int64, totalBytes int64, elapsedTime time.Duration, done bool)) (int64, error) {
402359
var uploadedBytes int64 = 0
403360
chunkID := 0

‎internal/turso/tursoServer_test.go‎

Lines changed: 0 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -1163,38 +1163,6 @@ func TestProgressReader_UpdatesTrackingFieldsCorrectly(t *testing.T) {
11631163
})
11641164
}
11651165

1166-
func TestUploadFileMultipartMVCC_RewritesFormatBytes(t *testing.T) {
1167-
mock := NewMockTursoServer()
1168-
mock.chunkSize = 16 // tiny chunks so bytes 18/19 land mid-stream
1169-
defer mock.Close()
1170-
1171-
content := make([]byte, 100)
1172-
for i := range content {
1173-
content[i] = byte(i)
1174-
}
1175-
content[18], content[19] = 2, 2 // WAL format version bytes
1176-
1177-
client := createTestClient(t, mock.URL)
1178-
testFile := createTestFileWithContent(t, content)
1179-
progress := NewProgressRecorder()
1180-
1181-
err := client.UploadFileMultipartMVCC(testFile, "", "", progress.Callback())
1182-
require.NoError(t, err)
1183-
1184-
uploaded := mock.GetAllChunkData()
1185-
require.Len(t, uploaded, len(content))
1186-
require.Equal(t, byte(255), uploaded[18])
1187-
require.Equal(t, byte(255), uploaded[19])
1188-
// everything else must be untouched
1189-
require.Equal(t, content[:18], uploaded[:18])
1190-
require.Equal(t, content[20:], uploaded[20:])
1191-
1192-
// the file on disk keeps its original bytes
1193-
onDisk, err := os.ReadFile(testFile)
1194-
require.NoError(t, err)
1195-
require.Equal(t, content, onDisk)
1196-
}
1197-
11981166
func TestUploadFileMultipart_DoesNotRewriteFormatBytes(t *testing.T) {
11991167
mock := NewMockTursoServer()
12001168
mock.chunkSize = 16

0 commit comments

Comments
 (0)