Fix RBF sync calls

rbf/db.go

@@ -294,6 +294,11 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error {
 		}
 	}
 
+	// Ensure WAL pages are fully copied & synced to DB file.
+	if err := db.file.Sync(); err != nil {
+		return fmt.Errorf("db file sync: %w", err)
+	}
+
 	// Remove WAL segments that have been checkpointed.
 	if maxCheckpointedWALID != 0 {
 		for _, segment := range segments {

rbf/rbf.go

@@ -648,3 +648,19 @@ func RowValues(b []uint64) []uint64 {
 // 	_, file, line, _ := runtime.Caller(skip + 1)
 // 	return fmt.Sprintf("%s:%d", file, line)
 // }
+
+// truncate truncates the file at path to sz bytes. File must exist.
+func truncate(path string, sz int64) error {
+	f, err := os.OpenFile(path, os.O_WRONLY, 0666)
+	if err != nil {
+		return fmt.Errorf("open file: %w", err)
+	}
+	defer f.Close()
+
+	if err := f.Truncate(sz); err != nil {
+		return fmt.Errorf("truncate: %w", err)
+	} else if err := f.Sync(); err != nil {
+		return fmt.Errorf("sync: %w", err)
+	}
+	return f.Close()
+}

rbf/wal.go

@@ -74,8 +74,8 @@ func (s *WALSegment) Open() (err error) {
 	s.PageN = int(sz / PageSize)
 	if sz%PageSize != 0 {
 		sz = int64(s.PageN * PageSize)
-		if err := os.Truncate(s.Path, sz); err != nil {
-			return fmt.Errorf("truncate wal segment file: %w", err)
+		if err := truncate(s.Path, sz); err != nil {
+			return fmt.Errorf("truncate wal file: %w", err)
 		}
 	}
 
@@ -229,7 +229,7 @@ func truncateWALAfter(segments []WALSegment, walID int64) ([]WALSegment, error)
 			newSegment := *segment
 			newSegment.PageN = int((walID - newSegment.MinWALID) + 1)
 
-			if err := os.Truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil {
+			if err := truncate(newSegment.Path, int64(newSegment.PageN)*PageSize); err != nil {
 				return segments, err
 			}
 			newSegments = append(newSegments, newSegment)