Remove journal on import (#213)

Author: benbjohnson

cmd/litefs-bench/main.go

@@ -62,7 +62,7 @@ func run(ctx context.Context) error {
 	defer func() { _ = db.Close() }()
 
 	if _, err := db.Exec(`PRAGMA busy_timeout = 5000`); err != nil {
-		return fmt.Errorf("set busy timeout")
+		return fmt.Errorf("set busy timeout: %w", err)
 	}
 
 	// Initialize cache.

cmd/litefs/import_test.go

@@ -48,6 +48,8 @@ func TestImportCommand_Create(t *testing.T) {
 
 // Ensure an existing database can be overwritten by an import.
 func TestImportCommand_Overwrite(t *testing.T) {
+	dir := t.TempDir()
+
 	// Generate a database on the regular file system.
 	dsn := filepath.Join(t.TempDir(), "db")
 	dbx := testingutil.OpenSQLDB(t, dsn)
@@ -60,7 +62,7 @@ func TestImportCommand_Overwrite(t *testing.T) {
 	}
 
 	// Run an LiteFS mount.
-	m0 := runMountCommand(t, newMountCommand(t, t.TempDir(), nil))
+	m0 := runMountCommand(t, newMountCommand(t, dir, nil))
 	waitForPrimary(t, m0)
 
 	// Generate data into the mount.
@@ -89,16 +91,37 @@ func TestImportCommand_Overwrite(t *testing.T) {
 		t.Fatal(err)
 	} else if got, want := y, 100; got != want {
 		t.Fatalf("y=%d, want %d", got, want)
+	} else if err := db.Close(); err != nil {
+		t.Fatal(err)
 	}
 
 	// Reconnect and verify correctness.
-	if err := db.Close(); err != nil {
-		t.Fatal(err)
-	}
 	db = testingutil.OpenSQLDB(t, filepath.Join(m0.Config.MountDir, "db"))
 	if err := db.QueryRow(`SELECT y FROM u`).Scan(&y); err != nil {
 		t.Fatal(err)
 	} else if got, want := y, 100; got != want {
 		t.Fatalf("y=%d, want %d", got, want)
 	}
+
+	// Add new transactions.
+	if _, err := db.Exec(`INSERT INTO u VALUES (200)`); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Close(); err != nil {
+		t.Fatal(err)
+	}
+
+	// Restart mount.
+	if err := m0.Close(); err != nil {
+		t.Fatal(err)
+	}
+	m0 = runMountCommand(t, newMountCommand(t, dir, m0))
+
+	db = testingutil.OpenSQLDB(t, filepath.Join(m0.Config.MountDir, "db"))
+	var sum int
+	if err := db.QueryRow(`SELECT SUM(y) FROM u`).Scan(&sum); err != nil {
+		t.Fatal(err)
+	} else if got, want := sum, 300; got != want {
+		t.Fatalf("sum=%d, want %d", got, want)
+	}
 }

db.go

@@ -288,6 +288,10 @@ func (db *DB) rollbackJournal(ctx context.Context) error {
 		return err
 	}
 
+	if err := os.Remove(db.JournalPath()); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -526,8 +530,8 @@ func (db *DB) WriteDatabase(f *os.File, data []byte, offset int64) error {
 	// Track dirty pages if we are using a rollback journal. This isn't
 	// necessary with the write-ahead log (WAL) since pages are appended
 	// instead of overwritten. We can determine the dirty set at commit-time.
+	pgno := uint32(offset/int64(db.pageSize)) + 1
 	if db.mode == DBModeRollback {
-		pgno := uint32(offset/int64(db.pageSize)) + 1
 		db.dirtyPageSet[pgno] = struct{}{}
 	}
 
@@ -1151,15 +1155,14 @@ func (db *DB) invalidateJournal(mode JournalMode) error {
 		f, err := os.OpenFile(db.JournalPath(), os.O_RDWR, 0666)
 		if err != nil && !os.IsNotExist(err) {
 			return fmt.Errorf("open journal: %w", err)
-		}
-		defer func() { _ = f.Close() }()
+		} else if err == nil {
+			defer func() { _ = f.Close() }()
 
-		if _, err := f.Write(make([]byte, SQLITE_JOURNAL_HEADER_SIZE)); err != nil {
-			return fmt.Errorf("clear journal header: %w", err)
-		} else if err := f.Sync(); err != nil {
-			return fmt.Errorf("sync journal: %w", err)
-		} else if err := f.Close(); err != nil {
-			return fmt.Errorf("close journal: %w", err)
+			if _, err := f.Write(make([]byte, SQLITE_JOURNAL_HEADER_SIZE)); err != nil {
+				return fmt.Errorf("clear journal header: %w", err)
+			} else if err := f.Sync(); err != nil {
+				return fmt.Errorf("sync journal: %w", err)
+			}
 		}
 
 	default:
@@ -1326,6 +1329,11 @@ func (db *DB) Import(ctx context.Context, r io.Reader) error {
 		return err
 	}
 
+	// Invalidate journal, if one exists.
+	if err := db.invalidateJournal(JournalModePersist); err != nil {
+		return fmt.Errorf("invalidate journal: %w", err)
+	}
+
 	// Truncate WAL, if it exists.
 	if _, err := os.Stat(db.WALPath()); err == nil {
 		if err := os.Truncate(db.WALPath(), 0); err != nil {
@@ -1397,8 +1405,9 @@ func (db *DB) importToLTX(ctx context.Context, r io.Reader) (Pos, error) {
 		pos.PostApplyChecksum ^= ltx.ChecksumPage(pgno, buf)
 	}
 
-	// Finish page block to compute checksum and then finish header block.
 	pos.PostApplyChecksum |= ltx.ChecksumFlag
+
+	// Finish page block to compute checksum and then finish header block.
 	enc.SetPostApplyChecksum(pos.PostApplyChecksum)
 	if err := enc.Close(); err != nil {
 		return Pos{}, fmt.Errorf("close ltx encoder: %s", err)

fuse/file_system_test.go

@@ -283,15 +283,11 @@ func TestFileSystem_ReadOnly(t *testing.T) {
 	_, err := db.Exec(`INSERT INTO t VALUES (100)`)
 
 	switch mode := testingutil.JournalMode(); mode {
-	case "delete":
+	case "delete", "persist", "truncate":
 		var e sqlite3.Error
 		if !errors.As(err, &e) || e.Code != sqlite3.ErrReadonly {
 			t.Fatalf("unexpected error: %s", err)
 		}
-	case "persist", "truncate":
-		if err == nil || err.Error() != `disk I/O error: permission denied` {
-			t.Fatalf("unexpected error: %s", err)
-		}
 	case "wal":
 		if err == nil || err.Error() != `disk I/O error` {
 			t.Fatalf("unexpected error: %s", err)