From eea1676ca458e76b11eb81cabba4fe0139bf66d1 Mon Sep 17 00:00:00 2001 From: AJ Roetker Date: Thu, 6 Nov 2025 04:20:40 +0000 Subject: [PATCH 1/5] (feat) add `vfs`: storage abstraction layer for Scorch `vfs` is a new storage abstraction layer that decouples Scorch's filesystem operations from the core indexing logic. This enables Scorch to use different storage backends (filesystem, S3, etc.) without modification. - Directory interface abstracting filesystem operations - FSDirectory: Drop-in replacement using local filesystem - Backward compatible with existing Scorch indexes - Allows custom storage backend implementations - Gets closer to enabling cloud-native and distributed architectures --- index/scorch/persister.go | 17 +- index/scorch/scorch.go | 38 +++- index/scorch/vfs/directory.go | 81 +++++++++ index/scorch/vfs/directory_test.go | 169 +++++++++++++++++ index/scorch/vfs/fs_directory.go | 214 ++++++++++++++++++++++ index/scorch/vfs/fs_directory_test.go | 253 ++++++++++++++++++++++++++ 6 files changed, 757 insertions(+), 15 deletions(-) create mode 100644 index/scorch/vfs/directory.go create mode 100644 index/scorch/vfs/directory_test.go create mode 100644 index/scorch/vfs/fs_directory.go create mode 100644 index/scorch/vfs/fs_directory_test.go diff --git a/index/scorch/persister.go b/index/scorch/persister.go index d92c3a85b..8e15f7020 100644 --- a/index/scorch/persister.go +++ b/index/scorch/persister.go @@ -1251,7 +1251,11 @@ func (s *Scorch) removeOldBoltSnapshots() (numRemoved int, err error) { } func (s *Scorch) maxSegmentIDOnDisk() (uint64, error) { - files, err := os.ReadDir(s.path) + if s.vfsDir == nil { + return 0, nil + } + + files, err := s.vfsDir.ReadDir(".") if err != nil { return 0, err } @@ -1280,18 +1284,23 @@ func (s *Scorch) removeOldZapFiles() error { return err } - files, err := os.ReadDir(s.path) + if s.vfsDir == nil { + return nil + } + + files, err := s.vfsDir.ReadDir(".") if err != nil { return err } s.rootLock.RLock() + defer s.rootLock.RUnlock() for _, f := range files { fname := f.Name() if filepath.Ext(fname) == ".zap" { if _, exists := liveFileNames[fname]; !exists && !s.ineligibleForRemoval[fname] && (s.copyScheduled[fname] <= 0) { - err := os.Remove(s.path + string(os.PathSeparator) + fname) + err := s.vfsDir.Remove(fname) if err != nil { log.Printf("got err removing file: %s, err: %v", fname, err) } @@ -1299,8 +1308,6 @@ func (s *Scorch) removeOldZapFiles() error { } } - s.rootLock.RUnlock() - return nil } diff --git a/index/scorch/scorch.go b/index/scorch/scorch.go index 83924978e..d80fd9ca7 100644 --- a/index/scorch/scorch.go +++ b/index/scorch/scorch.go @@ -24,6 +24,7 @@ import ( "time" "github.com/RoaringBitmap/roaring/v2" + "github.com/blevesearch/bleve/v2/index/scorch/vfs" "github.com/blevesearch/bleve/v2/registry" "github.com/blevesearch/bleve/v2/util" index "github.com/blevesearch/bleve_index_api" @@ -48,6 +49,10 @@ type Scorch struct { analysisQueue *index.AnalysisQueue path string + // vfsDir is the pluggable directory for segment storage + // If nil, falls back to filesystem operations at 'path' + vfsDir vfs.Directory + unsafeBatch bool rootLock sync.RWMutex @@ -145,6 +150,12 @@ func NewScorch(storeName string, } rv.root = &IndexSnapshot{parent: rv, refs: 1, creator: "NewScorch"} + + // Check if a custom VFS directory is provided + if dir, ok := config["vfsDirectory"].(vfs.Directory); ok { + rv.vfsDir = dir + } + ro, ok := config["read_only"].(bool) if ok { rv.readOnly = ro @@ -248,6 +259,15 @@ func (s *Scorch) openBolt() error { s.unsafeBatch = true } + // Initialize VFS directory if not already set + if s.vfsDir == nil && s.path != "" { + var err error + s.vfsDir, err = vfs.NewFSDirectory(s.path) + if err != nil { + return fmt.Errorf("failed to create VFS directory: %w", err) + } + } + rootBoltOpt := *bolt.DefaultOptions if s.readOnly { rootBoltOpt.ReadOnly = true @@ -600,19 +620,17 @@ func (s *Scorch) diskFileStats(rootSegmentPaths map[string]struct{}) (uint64, uint64, uint64, ) { var numFilesOnDisk, numBytesUsedDisk, numBytesOnDiskByRoot uint64 - if s.path != "" { - files, err := os.ReadDir(s.path) + if s.vfsDir != nil { + files, err := s.vfsDir.ReadDir(".") if err == nil { for _, f := range files { if !f.IsDir() { - if finfo, err := f.Info(); err == nil { - numBytesUsedDisk += uint64(finfo.Size()) - numFilesOnDisk++ - if rootSegmentPaths != nil { - fname := s.path + string(os.PathSeparator) + finfo.Name() - if _, fileAtRoot := rootSegmentPaths[fname]; fileAtRoot { - numBytesOnDiskByRoot += uint64(finfo.Size()) - } + numBytesUsedDisk += uint64(f.Size()) + numFilesOnDisk++ + if rootSegmentPaths != nil { + fname := s.path + string(os.PathSeparator) + f.Name() + if _, fileAtRoot := rootSegmentPaths[fname]; fileAtRoot { + numBytesOnDiskByRoot += uint64(f.Size()) } } } diff --git a/index/scorch/vfs/directory.go b/index/scorch/vfs/directory.go new file mode 100644 index 000000000..8f9685ec3 --- /dev/null +++ b/index/scorch/vfs/directory.go @@ -0,0 +1,81 @@ +// Copyright (c) 2025 Couchbase, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package vfs + +import ( + "io" + "io/fs" + "time" +) + +// Directory abstracts the filesystem operations required by Scorch. +// This interface enables Scorch to use different storage backends +// (local filesystem, S3, etc.) without modification. +// +// Implementations must be safe for concurrent use by multiple goroutines. +type Directory interface { + // Open opens the named file for reading. The caller must close the + // returned ReadCloser when done. + Open(name string) (io.ReadCloser, error) + + // Create creates or truncates the named file for writing. If the file + // already exists, it is truncated. The caller must close the returned + // WriteCloser when done. + Create(name string) (io.WriteCloser, error) + + // Remove removes the named file. + Remove(name string) error + + // Rename renames (moves) oldpath to newpath. If newpath already exists + // and is not a directory, Rename replaces it. + Rename(oldpath, newpath string) error + + // Stat returns FileInfo describing the named file. + Stat(name string) (FileInfo, error) + + // ReadDir reads the named directory and returns a list of directory entries. + ReadDir(name string) ([]FileInfo, error) + + // MkdirAll creates a directory named path, along with any necessary + // parents, and returns nil, or else returns an error. + MkdirAll(path string, perm fs.FileMode) error + + // Sync commits the current contents of the directory to stable storage. + // This is a hint that implementations can use to optimize durability. + Sync() error + + // Lock acquires an exclusive lock on the directory. This is used to + // prevent multiple processes from opening the same index simultaneously. + // Must be called before any other operations. + Lock() error + + // Unlock releases the lock acquired by Lock. + Unlock() error +} + +// FileInfo describes a file and is returned by Stat and ReadDir. +type FileInfo interface { + Name() string // base name of the file + Size() int64 // length in bytes + Mode() fs.FileMode // file mode bits + ModTime() time.Time // modification time + IsDir() bool // abbreviation for Mode().IsDir() +} + +// WriteCloser extends io.WriteCloser with a Sync method. +type WriteCloser interface { + io.WriteCloser + Sync() error +} diff --git a/index/scorch/vfs/directory_test.go b/index/scorch/vfs/directory_test.go new file mode 100644 index 000000000..4bf43c836 --- /dev/null +++ b/index/scorch/vfs/directory_test.go @@ -0,0 +1,169 @@ +// Copyright (c) 2025 Couchbase, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package vfs + +import ( + "io" + "testing" +) + +// directoryTestSuite runs a standard set of tests against any Directory implementation. +func directoryTestSuite(t *testing.T, dir Directory) { + t.Run("CreateAndRead", func(t *testing.T) { + testData := []byte("test data") + testFile := "test.dat" + + // Create and write + w, err := dir.Create(testFile) + if err != nil { + t.Fatalf("Failed to create file: %v", err) + } + + if _, err := w.Write(testData); err != nil { + t.Fatalf("Failed to write data: %v", err) + } + + if err := w.Close(); err != nil { + t.Fatalf("Failed to close writer: %v", err) + } + + // Read and verify + r, err := dir.Open(testFile) + if err != nil { + t.Fatalf("Failed to open file: %v", err) + } + defer r.Close() + + readData, err := io.ReadAll(r) + if err != nil { + t.Fatalf("Failed to read data: %v", err) + } + + if string(readData) != string(testData) { + t.Errorf("Data mismatch: got %q, want %q", readData, testData) + } + + // Clean up + if err := dir.Remove(testFile); err != nil { + t.Fatalf("Failed to remove file: %v", err) + } + }) + + t.Run("Stat", func(t *testing.T) { + testData := []byte("stat test") + testFile := "stat.dat" + + // Create file + w, err := dir.Create(testFile) + if err != nil { + t.Fatalf("Failed to create file: %v", err) + } + w.Write(testData) + w.Close() + + // Stat file + fi, err := dir.Stat(testFile) + if err != nil { + t.Fatalf("Failed to stat file: %v", err) + } + + if fi.Size() != int64(len(testData)) { + t.Errorf("Size mismatch: got %d, want %d", fi.Size(), len(testData)) + } + + if fi.IsDir() { + t.Error("Expected file, got directory") + } + + // Clean up + dir.Remove(testFile) + }) + + t.Run("Rename", func(t *testing.T) { + oldName := "old.dat" + newName := "new.dat" + testData := []byte("rename test") + + // Create file + w, err := dir.Create(oldName) + if err != nil { + t.Fatalf("Failed to create file: %v", err) + } + w.Write(testData) + w.Close() + + // Rename + if err := dir.Rename(oldName, newName); err != nil { + t.Fatalf("Failed to rename file: %v", err) + } + + // Verify new file exists and has correct content + r, err := dir.Open(newName) + if err != nil { + t.Fatalf("Failed to open renamed file: %v", err) + } + defer r.Close() + + readData, err := io.ReadAll(r) + if err != nil { + t.Fatalf("Failed to read renamed file: %v", err) + } + + if string(readData) != string(testData) { + t.Errorf("Data mismatch after rename") + } + + // Clean up + dir.Remove(newName) + }) + + t.Run("Remove", func(t *testing.T) { + testFile := "remove.dat" + + // Create file + w, err := dir.Create(testFile) + if err != nil { + t.Fatalf("Failed to create file: %v", err) + } + w.Write([]byte("remove test")) + w.Close() + + // Remove file + if err := dir.Remove(testFile); err != nil { + t.Fatalf("Failed to remove file: %v", err) + } + + // Verify file doesn't exist + if _, err := dir.Stat(testFile); err == nil { + t.Error("File still exists after remove") + } + }) +} + +func TestDirectoryCompliance_FSDirectory(t *testing.T) { + tmpDir := t.TempDir() + dir, err := NewFSDirectory(tmpDir) + if err != nil { + t.Fatalf("Failed to create FSDirectory: %v", err) + } + + // Lock directory for tests + if err := dir.Lock(); err != nil { + t.Fatalf("Failed to lock directory: %v", err) + } + defer dir.Unlock() + + directoryTestSuite(t, dir) +} diff --git a/index/scorch/vfs/fs_directory.go b/index/scorch/vfs/fs_directory.go new file mode 100644 index 000000000..d97faccc3 --- /dev/null +++ b/index/scorch/vfs/fs_directory.go @@ -0,0 +1,214 @@ +// Copyright (c) 2025 Couchbase, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package vfs + +import ( + "fmt" + "io" + "io/fs" + "os" + "path/filepath" + "sync" + "syscall" +) + +// FSDirectory is a Directory implementation that uses the local filesystem. +type FSDirectory struct { + basePath string + lockFile *os.File + mu sync.RWMutex +} + +// NewFSDirectory creates a new filesystem-based Directory at the given path. +func NewFSDirectory(path string) (*FSDirectory, error) { + absPath, err := filepath.Abs(path) + if err != nil { + return nil, fmt.Errorf("failed to get absolute path: %w", err) + } + + return &FSDirectory{ + basePath: absPath, + }, nil +} + +// Open opens the named file for reading. +func (d *FSDirectory) Open(name string) (io.ReadCloser, error) { + fullPath := d.FullPath(name) + return os.Open(fullPath) +} + +// Create creates or truncates the named file for writing. +func (d *FSDirectory) Create(name string) (io.WriteCloser, error) { + fullPath := d.FullPath(name) + + // Ensure parent directory exists + dir := filepath.Dir(fullPath) + if err := os.MkdirAll(dir, 0755); err != nil { + return nil, fmt.Errorf("failed to create parent directory: %w", err) + } + + f, err := os.Create(fullPath) + if err != nil { + return nil, err + } + + return &fsWriteCloser{File: f}, nil +} + +// Remove removes the named file. +func (d *FSDirectory) Remove(name string) error { + fullPath := d.FullPath(name) + return os.Remove(fullPath) +} + +// Rename renames (moves) oldpath to newpath. +func (d *FSDirectory) Rename(oldpath, newpath string) error { + oldFullPath := d.FullPath(oldpath) + newFullPath := d.FullPath(newpath) + + // Ensure parent directory of new path exists + dir := filepath.Dir(newFullPath) + if err := os.MkdirAll(dir, 0755); err != nil { + return fmt.Errorf("failed to create parent directory: %w", err) + } + + return os.Rename(oldFullPath, newFullPath) +} + +// Stat returns FileInfo describing the named file. +func (d *FSDirectory) Stat(name string) (FileInfo, error) { + fullPath := d.FullPath(name) + fi, err := os.Stat(fullPath) + if err != nil { + return nil, err + } + return &fsFileInfo{FileInfo: fi}, nil +} + +// ReadDir reads the named directory and returns a list of directory entries. +func (d *FSDirectory) ReadDir(name string) ([]FileInfo, error) { + fullPath := d.FullPath(name) + entries, err := os.ReadDir(fullPath) + if err != nil { + return nil, err + } + + result := make([]FileInfo, 0, len(entries)) + for _, entry := range entries { + info, err := entry.Info() + if err != nil { + continue // skip entries we can't stat + } + result = append(result, &fsFileInfo{FileInfo: info}) + } + return result, nil +} + +// MkdirAll creates a directory named path, along with any necessary parents. +func (d *FSDirectory) MkdirAll(path string, perm fs.FileMode) error { + fullPath := d.FullPath(path) + return os.MkdirAll(fullPath, perm) +} + +// Sync is a no-op for FSDirectory as file syncs happen on close. +func (d *FSDirectory) Sync() error { + // For filesystem directories, we don't need to do anything special here. + // Individual file syncs happen when files are closed. + return nil +} + +// Lock acquires an exclusive lock on the directory. +func (d *FSDirectory) Lock() error { + d.mu.Lock() + defer d.mu.Unlock() + + if d.lockFile != nil { + return fmt.Errorf("directory is already locked") + } + + lockPath := filepath.Join(d.basePath, "write.lock") + + // Ensure base directory exists + if err := os.MkdirAll(d.basePath, 0755); err != nil { + return fmt.Errorf("failed to create base directory: %w", err) + } + + f, err := os.OpenFile(lockPath, os.O_CREATE|os.O_RDWR, 0644) + if err != nil { + return fmt.Errorf("failed to create lock file: %w", err) + } + + // Try to acquire an exclusive lock (non-blocking) + err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) + if err != nil { + f.Close() + return fmt.Errorf("failed to acquire lock (another process may have the index open): %w", err) + } + + d.lockFile = f + return nil +} + +// Unlock releases the lock acquired by Lock. +func (d *FSDirectory) Unlock() error { + d.mu.Lock() + defer d.mu.Unlock() + + if d.lockFile == nil { + return nil // not locked + } + + // Release the lock + if err := syscall.Flock(int(d.lockFile.Fd()), syscall.LOCK_UN); err != nil { + return fmt.Errorf("failed to release lock: %w", err) + } + + // Close and remove the lock file + lockPath := d.lockFile.Name() + if err := d.lockFile.Close(); err != nil { + return fmt.Errorf("failed to close lock file: %w", err) + } + + // Try to remove the lock file (best effort, ignore errors) + _ = os.Remove(lockPath) + + d.lockFile = nil + return nil +} + +// FullPath returns the full filesystem path for a given name. +func (d *FSDirectory) FullPath(name string) string { + if filepath.IsAbs(name) { + return name + } + return filepath.Join(d.basePath, name) +} + +// fsWriteCloser wraps os.File to implement WriteCloser with Sync. +type fsWriteCloser struct { + *os.File +} + +func (w *fsWriteCloser) Sync() error { + return w.File.Sync() +} + +// fsFileInfo wraps os.FileInfo to implement our FileInfo interface. +type fsFileInfo struct { + os.FileInfo +} + +// Ensure FSDirectory implements Directory +var _ Directory = (*FSDirectory)(nil) diff --git a/index/scorch/vfs/fs_directory_test.go b/index/scorch/vfs/fs_directory_test.go new file mode 100644 index 000000000..9083db12c --- /dev/null +++ b/index/scorch/vfs/fs_directory_test.go @@ -0,0 +1,253 @@ +// Copyright (c) 2025 Couchbase, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package vfs + +import ( + "io" + "os" + "path/filepath" + "testing" +) + +func TestFSDirectory_BasicOperations(t *testing.T) { + // Create temporary directory + tmpDir := t.TempDir() + + // Create FSDirectory + dir, err := NewFSDirectory(tmpDir) + if err != nil { + t.Fatalf("Failed to create FSDirectory: %v", err) + } + + // Test Lock/Unlock + if err := dir.Lock(); err != nil { + t.Fatalf("Failed to lock directory: %v", err) + } + defer dir.Unlock() + + // Test Create and Write + testData := []byte("Hello, Firebug!") + testFile := "test.txt" + + w, err := dir.Create(testFile) + if err != nil { + t.Fatalf("Failed to create file: %v", err) + } + + if _, err := w.Write(testData); err != nil { + t.Fatalf("Failed to write data: %v", err) + } + + if err := w.Close(); err != nil { + t.Fatalf("Failed to close writer: %v", err) + } + + // Test Open and Read + r, err := dir.Open(testFile) + if err != nil { + t.Fatalf("Failed to open file: %v", err) + } + defer r.Close() + + readData, err := io.ReadAll(r) + if err != nil { + t.Fatalf("Failed to read data: %v", err) + } + + if string(readData) != string(testData) { + t.Errorf("Read data mismatch: got %q, want %q", readData, testData) + } + + // Test Stat + fi, err := dir.Stat(testFile) + if err != nil { + t.Fatalf("Failed to stat file: %v", err) + } + + if fi.Name() != testFile { + t.Errorf("File name mismatch: got %q, want %q", fi.Name(), testFile) + } + + if fi.Size() != int64(len(testData)) { + t.Errorf("File size mismatch: got %d, want %d", fi.Size(), len(testData)) + } + + // Test Rename + newName := "renamed.txt" + if err := dir.Rename(testFile, newName); err != nil { + t.Fatalf("Failed to rename file: %v", err) + } + + // Verify renamed file exists + if _, err := dir.Stat(newName); err != nil { + t.Fatalf("Failed to stat renamed file: %v", err) + } + + // Verify old file doesn't exist + if _, err := dir.Stat(testFile); !os.IsNotExist(err) { + t.Errorf("Old file still exists after rename") + } + + // Test Remove + if err := dir.Remove(newName); err != nil { + t.Fatalf("Failed to remove file: %v", err) + } + + // Verify file is removed + if _, err := dir.Stat(newName); !os.IsNotExist(err) { + t.Errorf("File still exists after remove") + } +} + +func TestFSDirectory_DirectoryOperations(t *testing.T) { + tmpDir := t.TempDir() + + dir, err := NewFSDirectory(tmpDir) + if err != nil { + t.Fatalf("Failed to create FSDirectory: %v", err) + } + + // Test MkdirAll + subdir := filepath.Join("a", "b", "c") + if err := dir.MkdirAll(subdir, 0755); err != nil { + t.Fatalf("Failed to create subdirectories: %v", err) + } + + // Create some test files + testFiles := []string{"file1.txt", "file2.txt", "file3.dat"} + for _, name := range testFiles { + w, err := dir.Create(name) + if err != nil { + t.Fatalf("Failed to create file %s: %v", name, err) + } + w.Write([]byte("test")) + w.Close() + } + + // Test ReadDir + entries, err := dir.ReadDir(".") + if err != nil { + t.Fatalf("Failed to read directory: %v", err) + } + + // Check that we have the expected number of entries + // (3 files + 1 subdirectory + possibly lock file) + if len(entries) < 4 { + t.Errorf("Expected at least 4 entries, got %d", len(entries)) + } + + // Verify our test files are in the list + found := make(map[string]bool) + for _, entry := range entries { + found[entry.Name()] = true + } + + for _, name := range testFiles { + if !found[name] { + t.Errorf("File %s not found in directory listing", name) + } + } +} + +func TestFSDirectory_Locking(t *testing.T) { + tmpDir := t.TempDir() + + dir1, err := NewFSDirectory(tmpDir) + if err != nil { + t.Fatalf("Failed to create first FSDirectory: %v", err) + } + + // Acquire lock with first directory + if err := dir1.Lock(); err != nil { + t.Fatalf("Failed to lock first directory: %v", err) + } + + // Try to acquire lock with second directory (should fail) + dir2, err := NewFSDirectory(tmpDir) + if err != nil { + t.Fatalf("Failed to create second FSDirectory: %v", err) + } + + if err := dir2.Lock(); err == nil { + t.Error("Expected lock to fail, but it succeeded") + dir2.Unlock() + } + + // Release first lock + if err := dir1.Unlock(); err != nil { + t.Fatalf("Failed to unlock first directory: %v", err) + } + + // Now second directory should be able to acquire lock + if err := dir2.Lock(); err != nil { + t.Fatalf("Failed to lock second directory after first was released: %v", err) + } + + if err := dir2.Unlock(); err != nil { + t.Fatalf("Failed to unlock second directory: %v", err) + } +} + +func TestFSDirectory_ConcurrentReads(t *testing.T) { + tmpDir := t.TempDir() + + dir, err := NewFSDirectory(tmpDir) + if err != nil { + t.Fatalf("Failed to create FSDirectory: %v", err) + } + + // Create test file + testFile := "concurrent.txt" + testData := []byte("Concurrent read test data") + + w, err := dir.Create(testFile) + if err != nil { + t.Fatalf("Failed to create file: %v", err) + } + w.Write(testData) + w.Close() + + // Perform concurrent reads + const numReaders = 10 + done := make(chan bool, numReaders) + + for i := 0; i < numReaders; i++ { + go func() { + defer func() { done <- true }() + + r, err := dir.Open(testFile) + if err != nil { + t.Errorf("Failed to open file: %v", err) + return + } + defer r.Close() + + data, err := io.ReadAll(r) + if err != nil { + t.Errorf("Failed to read file: %v", err) + return + } + + if string(data) != string(testData) { + t.Errorf("Data mismatch in concurrent read") + } + }() + } + + // Wait for all readers to complete + for i := 0; i < numReaders; i++ { + <-done + } +} From 3b6b4335bbdbbae763495705741c08b5c5f969b8 Mon Sep 17 00:00:00 2001 From: AJ Roetker Date: Mon, 10 Nov 2025 17:12:07 -0800 Subject: [PATCH 2/5] (bug) Fix flock issue on windows for new vfs abstraction --- index/scorch/merge.go | 57 ++++++-- index/scorch/persister.go | 130 +++++++++++------- index/scorch/scorch.go | 18 ++- index/scorch/segment_plugin.go | 40 ++++++ index/scorch/snapshot_index.go | 4 +- index/scorch/vfs/directory.go | 81 ----------- index/scorch/vfs/directory_test.go | 66 ++++++++- index/scorch/vfs/fs_directory.go | 32 +++-- index/scorch/vfs/fs_directory_lock_unix.go | 39 ++++++ index/scorch/vfs/fs_directory_lock_windows.go | 68 +++++++++ 10 files changed, 386 insertions(+), 149 deletions(-) delete mode 100644 index/scorch/vfs/directory.go create mode 100644 index/scorch/vfs/fs_directory_lock_unix.go create mode 100644 index/scorch/vfs/fs_directory_lock_windows.go diff --git a/index/scorch/merge.go b/index/scorch/merge.go index 9abcf2db6..4a13689bf 100644 --- a/index/scorch/merge.go +++ b/index/scorch/merge.go @@ -354,14 +354,26 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context, if len(segmentsToMerge) > 0 { filename = zapFileName(newSegmentID) s.markIneligibleForRemoval(filename) - path := s.path + string(os.PathSeparator) + filename fileMergeZapStartTime := time.Now() atomic.AddUint64(&s.stats.TotFileMergeZapBeg, 1) prevBytesReadTotal := cumulateBytesRead(segmentsToMerge) - newDocNums, _, err := s.segPlugin.Merge(segmentsToMerge, docsToDrop, path, - cw.cancelCh, s) + + // Try VFS-aware plugin first, fall back to legacy path-based + var newDocNums [][]uint64 + var err error + if vfsPlugin, ok := s.segPlugin.(SegmentPluginVFS); ok && s.vfsDir != nil { + // Use VFS-aware merge with relative filename + newDocNums, _, err = vfsPlugin.MergeVFS(s.vfsDir, filename, + segmentsToMerge, docsToDrop, cw.cancelCh, s) + } else { + // Legacy path-based merge + path := s.path + string(os.PathSeparator) + filename + newDocNums, _, err = s.segPlugin.Merge(segmentsToMerge, docsToDrop, path, + cw.cancelCh, s) + } + atomic.AddUint64(&s.stats.TotFileMergeZapEnd, 1) fileMergeZapTime := uint64(time.Since(fileMergeZapStartTime)) @@ -379,7 +391,15 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context, return fmt.Errorf("merging failed: %v", err) } - seg, err = s.segPlugin.Open(path) + // Open the newly merged segment + if vfsPlugin, ok := s.segPlugin.(SegmentPluginVFS); ok && s.vfsDir != nil { + // Use VFS-aware open with relative filename + seg, err = vfsPlugin.OpenVFS(s.vfsDir, filename) + } else { + // Legacy path-based open + path := s.path + string(os.PathSeparator) + filename + seg, err = s.segPlugin.Open(path) + } if err != nil { s.unmarkIneligibleForRemoval(filename) atomic.AddUint64(&s.stats.TotFileMergePlanTasksErr, 1) @@ -523,12 +543,23 @@ func (s *Scorch) mergeAndPersistInMemorySegments(snapshot *IndexSnapshot, defer wg.Done() newSegmentID := atomic.AddUint64(&s.nextSegmentID, 1) filename := zapFileName(newSegmentID) - path := s.path + string(os.PathSeparator) + filename // the newly merged segment is already flushed out to disk, just needs // to be opened using mmap. - newDocIDs, _, err := - s.segPlugin.Merge(segsBatch, dropsBatch, path, s.closeCh, s) + var newDocIDs [][]uint64 + var err error + + // Try VFS-aware plugin first, fall back to legacy path-based + if vfsPlugin, ok := s.segPlugin.(SegmentPluginVFS); ok && s.vfsDir != nil { + // Use VFS-aware merge with relative filename + newDocIDs, _, err = vfsPlugin.MergeVFS(s.vfsDir, filename, + segsBatch, dropsBatch, s.closeCh, s) + } else { + // Legacy path-based merge + path := s.path + string(os.PathSeparator) + filename + newDocIDs, _, err = s.segPlugin.Merge(segsBatch, dropsBatch, path, s.closeCh, s) + } + if err != nil { em.Lock() errs = append(errs, err) @@ -536,6 +567,7 @@ func (s *Scorch) mergeAndPersistInMemorySegments(snapshot *IndexSnapshot, atomic.AddUint64(&s.stats.TotMemMergeErr, 1) return } + // to prevent accidental cleanup of this newly created file, mark it // as ineligible for removal. this will be flipped back when the bolt // is updated - which is valid, since the snapshot updated in bolt is @@ -543,7 +575,16 @@ func (s *Scorch) mergeAndPersistInMemorySegments(snapshot *IndexSnapshot, s.markIneligibleForRemoval(filename) newMergedSegmentIDs[id] = newSegmentID newDocIDsSet[id] = newDocIDs - newMergedSegments[id], err = s.segPlugin.Open(path) + + // Open the newly merged segment + if vfsPlugin, ok := s.segPlugin.(SegmentPluginVFS); ok && s.vfsDir != nil { + // Use VFS-aware open with relative filename + newMergedSegments[id], err = vfsPlugin.OpenVFS(s.vfsDir, filename) + } else { + // Legacy path-based open + path := s.path + string(os.PathSeparator) + filename + newMergedSegments[id], err = s.segPlugin.Open(path) + } if err != nil { em.Lock() errs = append(errs, err) diff --git a/index/scorch/persister.go b/index/scorch/persister.go index 8e15f7020..fe8303623 100644 --- a/index/scorch/persister.go +++ b/index/scorch/persister.go @@ -34,6 +34,7 @@ import ( "github.com/RoaringBitmap/roaring/v2" "github.com/blevesearch/bleve/v2/util" index "github.com/blevesearch/bleve_index_api" + "github.com/blevesearch/bleve_index_api/vfs" segment "github.com/blevesearch/scorch_segment_api/v2" bolt "go.etcd.io/bbolt" ) @@ -554,59 +555,64 @@ func (s *Scorch) persistSnapshotMaybeMerge(snapshot *IndexSnapshot, po *persiste return true, nil } -func copyToDirectory(srcPath string, d index.Directory) (int64, error) { - if d == nil { - return 0, nil - } - - dest, err := d.GetWriter(filepath.Join("store", filepath.Base(srcPath))) +func copyBetweenDirectories(srcName string, srcDir vfs.Directory, dstDir vfs.Directory) error { + // Open source file from VFS + source, err := srcDir.Open(srcName) if err != nil { - return 0, fmt.Errorf("GetWriter err: %v", err) + return fmt.Errorf("VFS source open %s: %w", srcName, err) } + defer source.Close() - sourceFileStat, err := os.Stat(srcPath) + // Create destination file in VFS + dest, err := dstDir.Create(srcName) if err != nil { - return 0, err + return fmt.Errorf("VFS dest create %s: %w", srcName, err) } + defer dest.Close() - if !sourceFileStat.Mode().IsRegular() { - return 0, fmt.Errorf("%s is not a regular file", srcPath) + // Use 1MB buffer for better performance with remote storage + buf := make([]byte, 1024*1024) + if _, err := io.CopyBuffer(dest, source, buf); err != nil { + return fmt.Errorf("VFS copy %s: %w", srcName, err) } - source, err := os.Open(srcPath) - if err != nil { - return 0, err + // Critical: Sync destination for durability + if err := dest.Sync(); err != nil { + return fmt.Errorf("VFS dest sync %s: %w", srcName, err) } - defer source.Close() - defer dest.Close() - return io.Copy(dest, source) + + return nil } -func persistToDirectory(seg segment.UnpersistedSegment, d index.Directory, - path string, +func persistToDirectory(seg segment.UnpersistedSegment, d vfs.Directory, + name string, ) error { - if d == nil { - return seg.Persist(path) - } - + // Segments must implement io.WriterTo for VFS persistence sg, ok := seg.(io.WriterTo) if !ok { - return fmt.Errorf("no io.WriterTo segment implementation found") + return fmt.Errorf("segment doesn't implement io.WriterTo for VFS persistence") } - w, err := d.GetWriter(filepath.Join("store", filepath.Base(path))) + w, err := d.Create(name) if err != nil { - return err + return fmt.Errorf("VFS create %s: %w", name, err) } + defer w.Close() - _, err = sg.WriteTo(w) - w.Close() + if _, err := sg.WriteTo(w); err != nil { + return fmt.Errorf("segment write to %s: %w", name, err) + } + + // Critical: Sync before close for durability + if err := w.Sync(); err != nil { + return fmt.Errorf("VFS sync %s: %w", name, err) + } - return err + return nil } func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string, - segPlugin SegmentPlugin, exclude map[uint64]struct{}, d index.Directory) ( + segPlugin SegmentPlugin, exclude map[uint64]struct{}, d vfs.Directory) ( []string, map[uint64]string, error) { snapshotsBucket, err := tx.CreateBucketIfNotExists(util.BoltSnapshotsBucket) if err != nil { @@ -683,11 +689,9 @@ func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string, } switch seg := segmentSnapshot.segment.(type) { case segment.PersistedSegment: + // Persisted segments are already in the VFS directory + // No need to copy them - just record their metadata segPath := seg.Path() - _, err = copyToDirectory(segPath, d) - if err != nil { - return nil, nil, fmt.Errorf("segment: %s copy err: %v", segPath, err) - } filename := filepath.Base(segPath) err = snapshotSegmentBucket.Put(util.BoltPathKey, []byte(filename)) if err != nil { @@ -699,12 +703,21 @@ func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string, // restricts which in-memory segment to be persisted to disk) if _, ok := exclude[segmentSnapshot.id]; !ok { filename := zapFileName(segmentSnapshot.id) - path := filepath.Join(path, filename) - err := persistToDirectory(seg, d, path) + + // VFS is now mandatory for segment persistence + if d == nil { + return nil, nil, fmt.Errorf("VFS directory required for segment persistence") + } + + err := persistToDirectory(seg, d, filename) if err != nil { - return nil, nil, fmt.Errorf("segment: %s persist err: %v", path, err) + return nil, nil, fmt.Errorf("segment %s persist err: %v", filename, err) } - newSegmentPaths[segmentSnapshot.id] = path + + // Store the full path for backwards compatibility (used by segment opening code) + fullPath := filepath.Join(path, filename) + newSegmentPaths[segmentSnapshot.id] = fullPath + err = snapshotSegmentBucket.Put(util.BoltPathKey, []byte(filename)) if err != nil { return nil, nil, err @@ -768,7 +781,7 @@ func (s *Scorch) persistSnapshotDirect(snapshot *IndexSnapshot, exclude map[uint } }() - filenames, newSegmentPaths, err := prepareBoltSnapshot(snapshot, tx, s.path, s.segPlugin, exclude, nil) + filenames, newSegmentPaths, err := prepareBoltSnapshot(snapshot, tx, s.path, s.segPlugin, exclude, s.vfsDir) if err != nil { return err } @@ -793,9 +806,20 @@ func (s *Scorch) persistSnapshotDirect(snapshot *IndexSnapshot, exclude map[uint } }() for segmentID, path := range newSegmentPaths { - newSegments[segmentID], err = s.segPlugin.Open(path) - if err != nil { - return fmt.Errorf("error opening new segment at %s, %v", path, err) + // Try VFS-aware plugin first, fall back to legacy path-based + if vfsPlugin, ok := s.segPlugin.(SegmentPluginVFS); ok && s.vfsDir != nil { + // Use VFS-aware method with relative filename + filename := filepath.Base(path) + newSegments[segmentID], err = vfsPlugin.OpenVFS(s.vfsDir, filename) + if err != nil { + return fmt.Errorf("error opening segment %s via VFS, %v", filename, err) + } + } else { + // Legacy path-based opening + newSegments[segmentID], err = s.segPlugin.Open(path) + if err != nil { + return fmt.Errorf("error opening new segment at %s, %v", path, err) + } } } @@ -1004,10 +1028,24 @@ func (s *Scorch) loadSegment(segmentBucket *bolt.Bucket) (*SegmentSnapshot, erro if pathBytes == nil { return nil, fmt.Errorf("segment path missing") } - segmentPath := s.path + string(os.PathSeparator) + string(pathBytes) - seg, err := s.segPlugin.Open(segmentPath) - if err != nil { - return nil, fmt.Errorf("error opening bolt segment: %v", err) + filename := string(pathBytes) + + // Try VFS-aware plugin first, fall back to legacy path-based + var seg segment.Segment + var err error + if vfsPlugin, ok := s.segPlugin.(SegmentPluginVFS); ok && s.vfsDir != nil { + // Use VFS-aware method with relative filename + seg, err = vfsPlugin.OpenVFS(s.vfsDir, filename) + if err != nil { + return nil, fmt.Errorf("error opening segment %s via VFS: %v", filename, err) + } + } else { + // Legacy path-based opening + segmentPath := s.path + string(os.PathSeparator) + filename + seg, err = s.segPlugin.Open(segmentPath) + if err != nil { + return nil, fmt.Errorf("error opening bolt segment: %v", err) + } } rv := &SegmentSnapshot{ diff --git a/index/scorch/scorch.go b/index/scorch/scorch.go index d80fd9ca7..035f2f582 100644 --- a/index/scorch/scorch.go +++ b/index/scorch/scorch.go @@ -28,6 +28,7 @@ import ( "github.com/blevesearch/bleve/v2/registry" "github.com/blevesearch/bleve/v2/util" index "github.com/blevesearch/bleve_index_api" + apivfs "github.com/blevesearch/bleve_index_api/vfs" segment "github.com/blevesearch/scorch_segment_api/v2" bolt "go.etcd.io/bbolt" ) @@ -51,7 +52,7 @@ type Scorch struct { // vfsDir is the pluggable directory for segment storage // If nil, falls back to filesystem operations at 'path' - vfsDir vfs.Directory + vfsDir apivfs.Directory unsafeBatch bool @@ -152,7 +153,7 @@ func NewScorch(storeName string, rv.root = &IndexSnapshot{parent: rv, refs: 1, creator: "NewScorch"} // Check if a custom VFS directory is provided - if dir, ok := config["vfsDirectory"].(vfs.Directory); ok { + if dir, ok := config["vfsDirectory"].(apivfs.Directory); ok { rv.vfsDir = dir } @@ -281,10 +282,21 @@ func (s *Scorch) openBolt() error { return os.OpenFile(path, os.O_RDONLY, mode) } } else { + // Create base directory for BoltDB (local filesystem) if s.path != "" { err := os.MkdirAll(s.path, 0o700) if err != nil { - return err + return fmt.Errorf("create local directory: %w", err) + } + } + + // Ensure VFS directory exists for segments + // For FSDirectory, this is the same as the local directory + // For remote VFS (S3, etc.), this creates the necessary storage structure + if s.vfsDir != nil { + err := s.vfsDir.MkdirAll(".", 0o700) + if err != nil { + return fmt.Errorf("create VFS directory: %w", err) } } } diff --git a/index/scorch/segment_plugin.go b/index/scorch/segment_plugin.go index 790a8008a..fed185d43 100644 --- a/index/scorch/segment_plugin.go +++ b/index/scorch/segment_plugin.go @@ -20,6 +20,7 @@ import ( "github.com/RoaringBitmap/roaring/v2" "github.com/blevesearch/bleve/v2/geo" index "github.com/blevesearch/bleve_index_api" + "github.com/blevesearch/bleve_index_api/vfs" segment "github.com/blevesearch/scorch_segment_api/v2" zapv11 "github.com/blevesearch/zapx/v11" @@ -68,6 +69,45 @@ type SegmentPlugin interface { [][]uint64, uint64, error) } +// SegmentPluginVFS extends SegmentPlugin with VFS-aware methods. +// Plugins that implement this interface can work with the vfs.Directory +// abstraction, enabling support for remote storage backends (S3, GCS, etc.) +// in addition to local filesystem. +// +// The VFS-aware methods use relative file names instead of absolute paths, +// and all I/O operations go through the vfs.Directory interface. +// +// Implementations should maintain backwards compatibility by continuing to +// implement the base SegmentPlugin interface for legacy path-based operations. +type SegmentPluginVFS interface { + SegmentPlugin + + // OpenVFS attempts to open a segment file through the VFS directory + // and return the corresponding Segment. + // The name parameter is a relative filename (e.g., "000000000001.zap"), + // not an absolute path. + OpenVFS(dir vfs.Directory, name string) (segment.Segment, error) + + // MergeVFS takes a set of Segments and creates a new segment through + // the VFS directory at the specified relative name. + // This is the VFS-aware version of Merge() that works with the + // vfs.Directory abstraction instead of filesystem paths. + // + // Parameters: + // - dir: The VFS directory to write the merged segment to + // - name: Relative filename for the new segment (e.g., "000000000001.zap") + // - segments: Input segments to merge + // - drops: Bitmaps indicating which documents to drop during merge + // - closeCh: Channel to signal merge cancellation + // - s: Optional stats reporter for merge progress + // + // Returns: Same as Merge() - document mappings, bytes written, and error + MergeVFS(dir vfs.Directory, name string, + segments []segment.Segment, drops []*roaring.Bitmap, + closeCh chan struct{}, s segment.StatsReporter) ( + [][]uint64, uint64, error) +} + var supportedSegmentPlugins map[string]map[uint32]SegmentPlugin var defaultSegmentPlugin SegmentPlugin diff --git a/index/scorch/snapshot_index.go b/index/scorch/snapshot_index.go index c09a7db40..ab02402a4 100644 --- a/index/scorch/snapshot_index.go +++ b/index/scorch/snapshot_index.go @@ -1018,7 +1018,9 @@ func (is *IndexSnapshot) CopyTo(d index.Directory) error { return err } - _, _, err = prepareBoltSnapshot(is, tx, "", is.parent.segPlugin, nil, d) + // TODO: Update CopyTo to use vfs.Directory instead of index.Directory + // For now, pass nil to skip VFS path (backup functionality needs separate work) + _, _, err = prepareBoltSnapshot(is, tx, "", is.parent.segPlugin, nil, nil) if err != nil { _ = tx.Rollback() return fmt.Errorf("error backing up index snapshot: %v", err) diff --git a/index/scorch/vfs/directory.go b/index/scorch/vfs/directory.go deleted file mode 100644 index 8f9685ec3..000000000 --- a/index/scorch/vfs/directory.go +++ /dev/null @@ -1,81 +0,0 @@ -// Copyright (c) 2025 Couchbase, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package vfs - -import ( - "io" - "io/fs" - "time" -) - -// Directory abstracts the filesystem operations required by Scorch. -// This interface enables Scorch to use different storage backends -// (local filesystem, S3, etc.) without modification. -// -// Implementations must be safe for concurrent use by multiple goroutines. -type Directory interface { - // Open opens the named file for reading. The caller must close the - // returned ReadCloser when done. - Open(name string) (io.ReadCloser, error) - - // Create creates or truncates the named file for writing. If the file - // already exists, it is truncated. The caller must close the returned - // WriteCloser when done. - Create(name string) (io.WriteCloser, error) - - // Remove removes the named file. - Remove(name string) error - - // Rename renames (moves) oldpath to newpath. If newpath already exists - // and is not a directory, Rename replaces it. - Rename(oldpath, newpath string) error - - // Stat returns FileInfo describing the named file. - Stat(name string) (FileInfo, error) - - // ReadDir reads the named directory and returns a list of directory entries. - ReadDir(name string) ([]FileInfo, error) - - // MkdirAll creates a directory named path, along with any necessary - // parents, and returns nil, or else returns an error. - MkdirAll(path string, perm fs.FileMode) error - - // Sync commits the current contents of the directory to stable storage. - // This is a hint that implementations can use to optimize durability. - Sync() error - - // Lock acquires an exclusive lock on the directory. This is used to - // prevent multiple processes from opening the same index simultaneously. - // Must be called before any other operations. - Lock() error - - // Unlock releases the lock acquired by Lock. - Unlock() error -} - -// FileInfo describes a file and is returned by Stat and ReadDir. -type FileInfo interface { - Name() string // base name of the file - Size() int64 // length in bytes - Mode() fs.FileMode // file mode bits - ModTime() time.Time // modification time - IsDir() bool // abbreviation for Mode().IsDir() -} - -// WriteCloser extends io.WriteCloser with a Sync method. -type WriteCloser interface { - io.WriteCloser - Sync() error -} diff --git a/index/scorch/vfs/directory_test.go b/index/scorch/vfs/directory_test.go index 4bf43c836..fecc860cb 100644 --- a/index/scorch/vfs/directory_test.go +++ b/index/scorch/vfs/directory_test.go @@ -17,10 +17,12 @@ package vfs import ( "io" "testing" + + apivfs "github.com/blevesearch/bleve_index_api/vfs" ) // directoryTestSuite runs a standard set of tests against any Directory implementation. -func directoryTestSuite(t *testing.T, dir Directory) { +func directoryTestSuite(t *testing.T, dir apivfs.Directory) { t.Run("CreateAndRead", func(t *testing.T) { testData := []byte("test data") testFile := "test.dat" @@ -150,6 +152,68 @@ func directoryTestSuite(t *testing.T, dir Directory) { t.Error("File still exists after remove") } }) + + t.Run("OpenAt", func(t *testing.T) { + testData := []byte("random access test data with some content") + testFile := "openat.dat" + + // Create file + w, err := dir.Create(testFile) + if err != nil { + t.Fatalf("Failed to create file: %v", err) + } + if _, err := w.Write(testData); err != nil { + t.Fatalf("Failed to write data: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Failed to close writer: %v", err) + } + + // Open for random access + rac, err := dir.OpenAt(testFile) + if err != nil { + t.Fatalf("Failed to open file for random access: %v", err) + } + defer rac.Close() + + // Test reading at different positions + buf := make([]byte, 6) + + // Read from beginning + n, err := rac.ReadAt(buf, 0) + if err != nil && err != io.EOF { + t.Fatalf("Failed to read at position 0: %v", err) + } + if n != 6 { + t.Errorf("Expected to read 6 bytes, got %d", n) + } + if string(buf) != "random" { + t.Errorf("Data mismatch at position 0: got %q, want %q", buf, "random") + } + + // Read from middle + n, err = rac.ReadAt(buf, 7) + if err != nil && err != io.EOF { + t.Fatalf("Failed to read at position 7: %v", err) + } + if n != 6 { + t.Errorf("Expected to read 6 bytes, got %d", n) + } + if string(buf) != "access" { + t.Errorf("Data mismatch at position 7: got %q, want %q", buf, "access") + } + + // Test AsFd returns valid file descriptor + fd := rac.AsFd() + if fd == 0 { + t.Error("Expected valid file descriptor, got 0") + } + + // Clean up + if err := dir.Remove(testFile); err != nil { + t.Fatalf("Failed to remove file: %v", err) + } + }) } func TestDirectoryCompliance_FSDirectory(t *testing.T) { diff --git a/index/scorch/vfs/fs_directory.go b/index/scorch/vfs/fs_directory.go index d97faccc3..531cc21d4 100644 --- a/index/scorch/vfs/fs_directory.go +++ b/index/scorch/vfs/fs_directory.go @@ -21,7 +21,8 @@ import ( "os" "path/filepath" "sync" - "syscall" + + apivfs "github.com/blevesearch/bleve_index_api/vfs" ) // FSDirectory is a Directory implementation that uses the local filesystem. @@ -49,8 +50,19 @@ func (d *FSDirectory) Open(name string) (io.ReadCloser, error) { return os.Open(fullPath) } +// OpenAt opens the named file for random access reading. +// This is used for memory-mapped segments. +func (d *FSDirectory) OpenAt(name string) (apivfs.ReaderAtCloser, error) { + fullPath := d.FullPath(name) + f, err := os.Open(fullPath) + if err != nil { + return nil, err + } + return apivfs.NewFileReaderAtCloser(f), nil +} + // Create creates or truncates the named file for writing. -func (d *FSDirectory) Create(name string) (io.WriteCloser, error) { +func (d *FSDirectory) Create(name string) (apivfs.WriteCloser, error) { fullPath := d.FullPath(name) // Ensure parent directory exists @@ -88,7 +100,7 @@ func (d *FSDirectory) Rename(oldpath, newpath string) error { } // Stat returns FileInfo describing the named file. -func (d *FSDirectory) Stat(name string) (FileInfo, error) { +func (d *FSDirectory) Stat(name string) (apivfs.FileInfo, error) { fullPath := d.FullPath(name) fi, err := os.Stat(fullPath) if err != nil { @@ -98,14 +110,14 @@ func (d *FSDirectory) Stat(name string) (FileInfo, error) { } // ReadDir reads the named directory and returns a list of directory entries. -func (d *FSDirectory) ReadDir(name string) ([]FileInfo, error) { +func (d *FSDirectory) ReadDir(name string) ([]apivfs.FileInfo, error) { fullPath := d.FullPath(name) entries, err := os.ReadDir(fullPath) if err != nil { return nil, err } - result := make([]FileInfo, 0, len(entries)) + result := make([]apivfs.FileInfo, 0, len(entries)) for _, entry := range entries { info, err := entry.Info() if err != nil { @@ -151,7 +163,8 @@ func (d *FSDirectory) Lock() error { } // Try to acquire an exclusive lock (non-blocking) - err = syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB) + // Uses platform-specific implementation (flock on Unix, LockFileEx on Windows) + err = flock(f, true) if err != nil { f.Close() return fmt.Errorf("failed to acquire lock (another process may have the index open): %w", err) @@ -171,7 +184,8 @@ func (d *FSDirectory) Unlock() error { } // Release the lock - if err := syscall.Flock(int(d.lockFile.Fd()), syscall.LOCK_UN); err != nil { + // Uses platform-specific implementation (flock on Unix, UnlockFileEx on Windows) + if err := funlock(d.lockFile); err != nil { return fmt.Errorf("failed to release lock: %w", err) } @@ -210,5 +224,5 @@ type fsFileInfo struct { os.FileInfo } -// Ensure FSDirectory implements Directory -var _ Directory = (*FSDirectory)(nil) +// Ensure FSDirectory implements apivfs.Directory +var _ apivfs.Directory = (*FSDirectory)(nil) diff --git a/index/scorch/vfs/fs_directory_lock_unix.go b/index/scorch/vfs/fs_directory_lock_unix.go new file mode 100644 index 000000000..7ded31f63 --- /dev/null +++ b/index/scorch/vfs/fs_directory_lock_unix.go @@ -0,0 +1,39 @@ +//go:build !windows && !plan9 && !solaris && !aix && !android + +// Copyright (c) 2025 Couchbase, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package vfs + +import ( + "os" + "syscall" +) + +// flock acquires an advisory lock on a file using flock(2). +// This is the Unix/Linux/macOS implementation. +func flock(f *os.File, exclusive bool) error { + flag := syscall.LOCK_NB // Non-blocking + if exclusive { + flag |= syscall.LOCK_EX // Exclusive lock + } else { + flag |= syscall.LOCK_SH // Shared lock + } + return syscall.Flock(int(f.Fd()), flag) +} + +// funlock releases the advisory lock on a file. +func funlock(f *os.File) error { + return syscall.Flock(int(f.Fd()), syscall.LOCK_UN) +} diff --git a/index/scorch/vfs/fs_directory_lock_windows.go b/index/scorch/vfs/fs_directory_lock_windows.go new file mode 100644 index 000000000..8a4122af1 --- /dev/null +++ b/index/scorch/vfs/fs_directory_lock_windows.go @@ -0,0 +1,68 @@ +//go:build windows + +// Copyright (c) 2025 Couchbase, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package vfs + +import ( + "os" + + "golang.org/x/sys/windows" +) + +// flock acquires an advisory lock on a file using LockFileEx. +// This is the Windows implementation. +// Following the bbolt pattern: uses byte-range lock at offset -1..0 +func flock(f *os.File, exclusive bool) error { + // Flags for immediate failure if lock cannot be acquired + flags := uint32(windows.LOCKFILE_FAIL_IMMEDIATELY) + if exclusive { + flags |= windows.LOCKFILE_EXCLUSIVE_LOCK + } + + // Use byte-range -1..0 as the lock range (bbolt pattern) + // This avoids conflicts with actual file content + var m1 uint32 = (1 << 32) - 1 // -1 in a uint32 + + err := windows.LockFileEx( + windows.Handle(f.Fd()), + flags, + 0, // reserved, must be 0 + 1, // number of bytes to lock (low DWORD) + 0, // number of bytes to lock (high DWORD) + &windows.Overlapped{ + Offset: m1, + OffsetHigh: m1, + }, + ) + + return err +} + +// funlock releases the advisory lock on a file. +func funlock(f *os.File) error { + var m1 uint32 = (1 << 32) - 1 + + return windows.UnlockFileEx( + windows.Handle(f.Fd()), + 0, // reserved, must be 0 + 1, // number of bytes to unlock (low DWORD) + 0, // number of bytes to unlock (high DWORD) + &windows.Overlapped{ + Offset: m1, + OffsetHigh: m1, + }, + ) +} From 300307671830ec1f6d8eb7f01f2962057c5ead20 Mon Sep 17 00:00:00 2001 From: AJ Roetker Date: Tue, 11 Nov 2025 12:28:30 -0800 Subject: [PATCH 3/5] Update go.mod to point to bleve_index_api and zapx for VFS interfaces --- go.mod | 14 ++++++++++---- go.sum | 16 ++++++++-------- 2 files changed, 18 insertions(+), 12 deletions(-) diff --git a/go.mod b/go.mod index a758308b2..a98ddf648 100644 --- a/go.mod +++ b/go.mod @@ -7,14 +7,14 @@ toolchain go1.23.9 require ( github.com/RoaringBitmap/roaring/v2 v2.4.5 github.com/bits-and-blooms/bitset v1.22.0 - github.com/blevesearch/bleve_index_api v1.2.10 + github.com/blevesearch/bleve_index_api v1.2.11 github.com/blevesearch/geo v0.2.4 - github.com/blevesearch/go-faiss v1.0.25 + github.com/blevesearch/go-faiss v1.0.26 github.com/blevesearch/go-metrics v0.0.0-20201227073835-cf1acfcdf475 github.com/blevesearch/go-porterstemmer v1.0.3 github.com/blevesearch/goleveldb v1.0.1 github.com/blevesearch/gtreap v0.1.1 - github.com/blevesearch/scorch_segment_api/v2 v2.3.12 + github.com/blevesearch/scorch_segment_api/v2 v2.3.13 github.com/blevesearch/segment v0.9.1 github.com/blevesearch/snowball v0.6.1 github.com/blevesearch/snowballstem v0.9.0 @@ -30,6 +30,7 @@ require ( github.com/couchbase/moss v0.2.0 github.com/spf13/cobra v1.8.1 go.etcd.io/bbolt v1.4.0 + golang.org/x/sys v0.29.0 golang.org/x/text v0.8.0 google.golang.org/protobuf v1.36.6 ) @@ -42,5 +43,10 @@ require ( github.com/json-iterator/go v0.0.0-20171115153421-f7279a603ede // indirect github.com/mschoch/smat v0.2.0 // indirect github.com/spf13/pflag v1.0.6 // indirect - golang.org/x/sys v0.29.0 // indirect ) + +// Use bleve_index_api branch with VFS support +replace github.com/blevesearch/bleve_index_api => github.com/ajroetker/bleve_index_api v0.0.0-20251111010750-7b3692d79f01 + +// Use zapx branch with VFS support +replace github.com/blevesearch/zapx/v16 => github.com/ajroetker/zapx/v16 v16.0.0-20251111202718-67691ba74877 diff --git a/go.sum b/go.sum index 5567fce8e..fe6c1fcba 100644 --- a/go.sum +++ b/go.sum @@ -1,14 +1,16 @@ github.com/RoaringBitmap/roaring/v2 v2.4.5 h1:uGrrMreGjvAtTBobc0g5IrW1D5ldxDQYe2JW2gggRdg= github.com/RoaringBitmap/roaring/v2 v2.4.5/go.mod h1:FiJcsfkGje/nZBZgCu0ZxCPOKD/hVXDS2dXi7/eUFE0= +github.com/ajroetker/bleve_index_api v0.0.0-20251111010750-7b3692d79f01 h1:SbGoS4vY5GDtDwxKy4iT+s0LsEBQMKd/FNRwTCnWssI= +github.com/ajroetker/bleve_index_api v0.0.0-20251111010750-7b3692d79f01/go.mod h1:rKQDl4u51uwafZxFrPD1R7xFOwKnzZW7s/LSeK4lgo0= +github.com/ajroetker/zapx/v16 v16.0.0-20251111202718-67691ba74877 h1:LdZ2GwvNsTYSW9jyBwdGteL4wNisp4j7P8sEmehSU0k= +github.com/ajroetker/zapx/v16 v16.0.0-20251111202718-67691ba74877/go.mod h1:7y0yPdM9JLW29eRvgtmgaxujU4t0CUfzo1sFvP1lLss= github.com/bits-and-blooms/bitset v1.12.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= github.com/bits-and-blooms/bitset v1.22.0 h1:Tquv9S8+SGaS3EhyA+up3FXzmkhxPGjQQCkcs2uw7w4= github.com/bits-and-blooms/bitset v1.22.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= -github.com/blevesearch/bleve_index_api v1.2.10 h1:FMFmZCmTX6PdoLLvwUnKF2RsmILFFwO3h0WPevXY9fE= -github.com/blevesearch/bleve_index_api v1.2.10/go.mod h1:rKQDl4u51uwafZxFrPD1R7xFOwKnzZW7s/LSeK4lgo0= github.com/blevesearch/geo v0.2.4 h1:ECIGQhw+QALCZaDcogRTNSJYQXRtC8/m8IKiA706cqk= github.com/blevesearch/geo v0.2.4/go.mod h1:K56Q33AzXt2YExVHGObtmRSFYZKYGv0JEN5mdacJJR8= -github.com/blevesearch/go-faiss v1.0.25 h1:lel1rkOUGbT1CJ0YgzKwC7k+XH0XVBHnCVWahdCXk4U= -github.com/blevesearch/go-faiss v1.0.25/go.mod h1:OMGQwOaRRYxrmeNdMrXJPvVx8gBnvE5RYrr0BahNnkk= +github.com/blevesearch/go-faiss v1.0.26 h1:4dRLolFgjPyjkaXwff4NfbZFdE/dfywbzDqporeQvXI= +github.com/blevesearch/go-faiss v1.0.26/go.mod h1:OMGQwOaRRYxrmeNdMrXJPvVx8gBnvE5RYrr0BahNnkk= github.com/blevesearch/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:kDy+zgJFJJoJYBvdfBSiZYBbdsUL0XcjHYWezpQBGPA= github.com/blevesearch/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:9eJDeqxJ3E7WnLebQUlPD7ZjSce7AnDb9vjGmMCbD0A= github.com/blevesearch/go-porterstemmer v1.0.3 h1:GtmsqID0aZdCSNiY8SkuPJ12pD4jI+DdXTAn4YRcHCo= @@ -20,8 +22,8 @@ github.com/blevesearch/gtreap v0.1.1/go.mod h1:QaQyDRAT51sotthUWAH4Sj08awFSSWzgY github.com/blevesearch/mmap-go v1.0.2/go.mod h1:ol2qBqYaOUsGdm7aRMRrYGgPvnwLe6Y+7LMvAB5IbSA= github.com/blevesearch/mmap-go v1.0.4 h1:OVhDhT5B/M1HNPpYPBKIEJaD0F3Si+CrEKULGCDPWmc= github.com/blevesearch/mmap-go v1.0.4/go.mod h1:EWmEAOmdAS9z/pi/+Toxu99DnsbhG1TIxUoRmJw/pSs= -github.com/blevesearch/scorch_segment_api/v2 v2.3.12 h1:GGZc2qwbyRBwtckPPkHkLyXw64mmsLJxdturBI1cM+c= -github.com/blevesearch/scorch_segment_api/v2 v2.3.12/go.mod h1:JBRGAneqgLSI2+jCNjtwMqp2B7EBF3/VUzgDPIU33MM= +github.com/blevesearch/scorch_segment_api/v2 v2.3.13 h1:ZPjv/4VwWvHJZKeMSgScCapOy8+DdmsmRyLmSB88UoY= +github.com/blevesearch/scorch_segment_api/v2 v2.3.13/go.mod h1:ENk2LClTehOuMS8XzN3UxBEErYmtwkE7MAArFTXs9Vc= github.com/blevesearch/segment v0.9.1 h1:+dThDy+Lvgj5JMxhmOVlgFfkUtZV2kw49xax4+jTfSU= github.com/blevesearch/segment v0.9.1/go.mod h1:zN21iLm7+GnBHWTao9I+Au/7MBiL8pPFtJBJTsk6kQw= github.com/blevesearch/snowball v0.6.1 h1:cDYjn/NCH+wwt2UdehaLpr2e4BwLIjN4V/TdLsL+B5A= @@ -44,8 +46,6 @@ github.com/blevesearch/zapx/v14 v14.4.2 h1:2SGHakVKd+TrtEqpfeq8X+So5PShQ5nW6GNxT github.com/blevesearch/zapx/v14 v14.4.2/go.mod h1:rz0XNb/OZSMjNorufDGSpFpjoFKhXmppH9Hi7a877D8= github.com/blevesearch/zapx/v15 v15.4.2 h1:sWxpDE0QQOTjyxYbAVjt3+0ieu8NCE0fDRaFxEsp31k= github.com/blevesearch/zapx/v15 v15.4.2/go.mod h1:1pssev/59FsuWcgSnTa0OeEpOzmhtmr/0/11H0Z8+Nw= -github.com/blevesearch/zapx/v16 v16.2.6 h1:OHuUl2GhM+FpBq9RwNsJ4k/QodqbMMHoQEgn/IHYpu8= -github.com/blevesearch/zapx/v16 v16.2.6/go.mod h1:cuAPB+YoIyRngNhno1S1GPr9SfMk+x/SgAHBLXSIq3k= github.com/couchbase/ghistogram v0.1.0 h1:b95QcQTCzjTUocDXp/uMgSNQi8oj1tGwnJ4bODWZnps= github.com/couchbase/ghistogram v0.1.0/go.mod h1:s1Jhy76zqfEecpNWJfWUiKZookAFaiGOEoyzgHt9i7k= github.com/couchbase/moss v0.2.0 h1:VCYrMzFwEryyhRSeI+/b3tRBSeTpi/8gn5Kf6dxqn+o= From 6149e5e86d22391edd4c288275980e81dd1a4350 Mon Sep 17 00:00:00 2001 From: AJ Roetker Date: Tue, 11 Nov 2025 15:47:46 -0800 Subject: [PATCH 4/5] Implement VFS index copying and fix merge cancellation Complete the VFS abstraction work by implementing CopyTo for online index backups and fixing a bug where merge cancellation wasn't checked between tasks. - Implement snapshot_index.go CopyTo to copy bolt + segments via VFS - Add cancellation check in merge loop between tasks --- go.mod | 4 +- go.sum | 4 +- index/scorch/merge.go | 7 ++ index/scorch/persister.go | 59 +++++++++++++++- index/scorch/rollback_test.go | 8 ++- index/scorch/snapshot_index.go | 116 ++++++++++++++++++++++++------- index/scorch/vfs/fs_directory.go | 17 ++++- index_test.go | 9 ++- 8 files changed, 189 insertions(+), 35 deletions(-) diff --git a/go.mod b/go.mod index 6a2482251..3edab3bb9 100644 --- a/go.mod +++ b/go.mod @@ -48,5 +48,5 @@ require ( // Use bleve_index_api branch with VFS support replace github.com/blevesearch/bleve_index_api => github.com/ajroetker/bleve_index_api v0.0.0-20251111010750-7b3692d79f01 -// Use zapx branch with VFS support -replace github.com/blevesearch/zapx/v16 => github.com/ajroetker/zapx/v16 v16.0.0-20251111202718-67691ba74877 +// Use VFS-enabled zapx from ajroetker fork +replace github.com/blevesearch/zapx/v16 => github.com/ajroetker/zapx/v16 v16.0.0-20251111234330-70822381ed85 diff --git a/go.sum b/go.sum index fe6c1fcba..017fd8000 100644 --- a/go.sum +++ b/go.sum @@ -2,8 +2,8 @@ github.com/RoaringBitmap/roaring/v2 v2.4.5 h1:uGrrMreGjvAtTBobc0g5IrW1D5ldxDQYe2 github.com/RoaringBitmap/roaring/v2 v2.4.5/go.mod h1:FiJcsfkGje/nZBZgCu0ZxCPOKD/hVXDS2dXi7/eUFE0= github.com/ajroetker/bleve_index_api v0.0.0-20251111010750-7b3692d79f01 h1:SbGoS4vY5GDtDwxKy4iT+s0LsEBQMKd/FNRwTCnWssI= github.com/ajroetker/bleve_index_api v0.0.0-20251111010750-7b3692d79f01/go.mod h1:rKQDl4u51uwafZxFrPD1R7xFOwKnzZW7s/LSeK4lgo0= -github.com/ajroetker/zapx/v16 v16.0.0-20251111202718-67691ba74877 h1:LdZ2GwvNsTYSW9jyBwdGteL4wNisp4j7P8sEmehSU0k= -github.com/ajroetker/zapx/v16 v16.0.0-20251111202718-67691ba74877/go.mod h1:7y0yPdM9JLW29eRvgtmgaxujU4t0CUfzo1sFvP1lLss= +github.com/ajroetker/zapx/v16 v16.0.0-20251111234330-70822381ed85 h1:pY9/pLIPBX6JMYI9DzultU8yybMBVaqzN8ceU/dQurQ= +github.com/ajroetker/zapx/v16 v16.0.0-20251111234330-70822381ed85/go.mod h1:7y0yPdM9JLW29eRvgtmgaxujU4t0CUfzo1sFvP1lLss= github.com/bits-and-blooms/bitset v1.12.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= github.com/bits-and-blooms/bitset v1.22.0 h1:Tquv9S8+SGaS3EhyA+up3FXzmkhxPGjQQCkcs2uw7w4= github.com/bits-and-blooms/bitset v1.22.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8= diff --git a/index/scorch/merge.go b/index/scorch/merge.go index 4a13689bf..c64315740 100644 --- a/index/scorch/merge.go +++ b/index/scorch/merge.go @@ -310,6 +310,13 @@ func (s *Scorch) planMergeAtSnapshot(ctx context.Context, go cw.listen() for _, task := range resultMergePlan.Tasks { + // Check if context was cancelled before starting next task + select { + case <-cw.cancelCh: + return segment.ErrClosed + default: + } + if len(task.Segments) == 0 { atomic.AddUint64(&s.stats.TotFileMergePlanTasksSegmentsEmpty, 1) continue diff --git a/index/scorch/persister.go b/index/scorch/persister.go index fe8303623..ab69177bc 100644 --- a/index/scorch/persister.go +++ b/index/scorch/persister.go @@ -599,9 +599,13 @@ func persistToDirectory(seg segment.UnpersistedSegment, d vfs.Directory, } defer w.Close() - if _, err := sg.WriteTo(w); err != nil { + n, err := sg.WriteTo(w) + if err != nil { return fmt.Errorf("segment write to %s: %w", name, err) } + if n == 0 { + return fmt.Errorf("segment write to %s produced 0 bytes", name) + } // Critical: Sync before close for durability if err := w.Sync(); err != nil { @@ -611,6 +615,41 @@ func persistToDirectory(seg segment.UnpersistedSegment, d vfs.Directory, return nil } +// copySegmentFile copies a segment file from source VFS to destination VFS. +func copySegmentFile(src, dst vfs.Directory, filename string) error { + return copySegmentFileWithPath(src, dst, filename, filename) +} + +// copySegmentFileWithPath copies a segment file from source VFS to destination VFS, +// allowing different paths for source and destination. +func copySegmentFileWithPath(src, dst vfs.Directory, srcFilename, dstFilename string) error { + // Open source file for reading + r, err := src.Open(srcFilename) + if err != nil { + return fmt.Errorf("failed to open source file %s: %w", srcFilename, err) + } + defer r.Close() + + // Create destination file for writing + w, err := dst.Create(dstFilename) + if err != nil { + return fmt.Errorf("failed to create dest file %s: %w", dstFilename, err) + } + defer w.Close() + + // Copy the file contents + if _, err := io.Copy(w, r); err != nil { + return fmt.Errorf("failed to copy data from %s to %s: %w", srcFilename, dstFilename, err) + } + + // Sync to ensure durability + if err := w.Sync(); err != nil { + return fmt.Errorf("failed to sync dest file %s: %w", dstFilename, err) + } + + return nil +} + func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string, segPlugin SegmentPlugin, exclude map[uint64]struct{}, d vfs.Directory) ( []string, map[uint64]string, error) { @@ -689,10 +728,24 @@ func prepareBoltSnapshot(snapshot *IndexSnapshot, tx *bolt.Tx, path string, } switch seg := segmentSnapshot.segment.(type) { case segment.PersistedSegment: - // Persisted segments are already in the VFS directory - // No need to copy them - just record their metadata + // Persisted segments are already in the source VFS directory segPath := seg.Path() filename := filepath.Base(segPath) + + // If destination VFS is provided and different from source, copy the segment file + if d != nil && snapshot.parent != nil && snapshot.parent.vfsDir != nil && d != snapshot.parent.vfsDir { + // Source and destination are different - copy the segment file + // If path is specified (e.g. "store"), prepend it to the filename for destination + destFilename := filename + if path != "" { + destFilename = filepath.Join(path, filename) + } + err := copySegmentFileWithPath(snapshot.parent.vfsDir, d, filename, destFilename) + if err != nil { + return nil, nil, fmt.Errorf("failed to copy segment %s: %w", filename, err) + } + } + err = snapshotSegmentBucket.Put(util.BoltPathKey, []byte(filename)) if err != nil { return nil, nil, err diff --git a/index/scorch/rollback_test.go b/index/scorch/rollback_test.go index c0facfedf..154bd941e 100644 --- a/index/scorch/rollback_test.go +++ b/index/scorch/rollback_test.go @@ -23,6 +23,7 @@ import ( "time" "github.com/blevesearch/bleve/v2/document" + "github.com/blevesearch/bleve/v2/index/scorch/vfs" index "github.com/blevesearch/bleve_index_api" ) @@ -545,7 +546,12 @@ func TestBackupRacingWithPurge(t *testing.T) { }() // if the latest snapshot was purged, the following will return error - err = copyReader.CopyTo(testFSDirector(backupidxConfig["path"].(string))) + backupPath := backupidxConfig["path"].(string) + backupVFSDir, err := vfs.NewFSDirectory(filepath.Join(backupPath, "store")) + if err != nil { + t.Fatalf("error creating backup VFS directory: %v", err) + } + err = copyReader.CopyTo(backupVFSDir) if err != nil { t.Fatalf("error copying the index: %v", err) } diff --git a/index/scorch/snapshot_index.go b/index/scorch/snapshot_index.go index 3d05ddea8..807d29c63 100644 --- a/index/scorch/snapshot_index.go +++ b/index/scorch/snapshot_index.go @@ -19,6 +19,7 @@ import ( "context" "encoding/binary" "fmt" + "io" "os" "path/filepath" "reflect" @@ -30,6 +31,7 @@ import ( "github.com/RoaringBitmap/roaring/v2" "github.com/blevesearch/bleve/v2/document" index "github.com/blevesearch/bleve_index_api" + "github.com/blevesearch/bleve_index_api/vfs" segment "github.com/blevesearch/scorch_segment_api/v2" "github.com/blevesearch/vellum" lev "github.com/blevesearch/vellum/levenshtein" @@ -992,48 +994,112 @@ OUTER: } func (is *IndexSnapshot) CopyTo(d index.Directory) error { - // get the root bolt file. - w, err := d.GetWriter(filepath.Join("store", "root.bolt")) - if err != nil || w == nil { - return fmt.Errorf("failed to create the root.bolt file, err: %v", err) + // Convert index.Directory to vfs.Directory if possible + // For backwards compatibility, we need to support both old-style index.Directory + // and new VFS-aware destinations + var destVFS vfs.Directory + + // Check if destination implements vfs.Directory + if vfsDir, ok := d.(vfs.Directory); ok { + destVFS = vfsDir + } else { + // Legacy path: destination is old-style index.Directory (e.g., FileSystemDirectory) + // We need to extract the base path and create a VFS directory from it + // This is for backwards compatibility with existing code + return fmt.Errorf("CopyTo requires a vfs.Directory implementation; legacy index.Directory is no longer supported for copying segment files") } - rootFile, ok := w.(*os.File) - if !ok { - return fmt.Errorf("invalid root.bolt file found") + + // Create the store subdirectory in the destination + // The index structure has metadata at the root and scorch data in "store/" + storeDir := "store" + + // get the root bolt file writer in the store subdirectory + w, err := destVFS.Create(filepath.Join(storeDir, "root.bolt")) + if err != nil { + return fmt.Errorf("failed to create the root.bolt file: %w", err) } - copyBolt, err := bolt.Open(rootFile.Name(), 0o600, nil) + // For BoltDB, we need a real file path, so we need to use a temporary file approach + // Create a temporary file, populate it with BoltDB data, then copy to VFS + tmpFile, err := os.CreateTemp("", "backup-root-*.bolt") if err != nil { - return err + w.Close() + return fmt.Errorf("failed to create temp bolt file: %w", err) } - defer func() { + tmpPath := tmpFile.Name() + tmpFile.Close() + defer os.Remove(tmpPath) + + copyBolt, err := bolt.Open(tmpPath, 0o600, nil) + if err != nil { w.Close() - if cerr := copyBolt.Close(); cerr != nil && err == nil { - err = cerr + return fmt.Errorf("failed to open temp bolt: %w", err) + } + + var txErr error + func() { + defer copyBolt.Close() + + // start a write transaction + tx, err := copyBolt.Begin(true) + if err != nil { + txErr = err + return + } + + // Prepare the snapshot in BoltDB and copy segment files to destination VFS + // Pass "store" as the path so segments are copied to the store subdirectory + _, _, err = prepareBoltSnapshot(is, tx, storeDir, is.parent.segPlugin, nil, destVFS) + if err != nil { + _ = tx.Rollback() + txErr = fmt.Errorf("error backing up index snapshot: %w", err) + return + } + + // commit bolt data + err = tx.Commit() + if err != nil { + txErr = fmt.Errorf("error commit tx to backup root bolt: %w", err) + return + } + + err = copyBolt.Sync() + if err != nil { + txErr = fmt.Errorf("error syncing bolt: %w", err) + return } }() - // start a write transaction - tx, err := copyBolt.Begin(true) - if err != nil { - return err + if txErr != nil { + w.Close() + return txErr } - // TODO: Update CopyTo to use vfs.Directory instead of index.Directory - // For now, pass nil to skip VFS path (backup functionality needs separate work) - _, _, err = prepareBoltSnapshot(is, tx, "", is.parent.segPlugin, nil, nil) + // Now copy the temp bolt file to destination VFS + tmpBolt, err := os.Open(tmpPath) if err != nil { - _ = tx.Rollback() - return fmt.Errorf("error backing up index snapshot: %v", err) + w.Close() + return fmt.Errorf("failed to open temp bolt for reading: %w", err) } + defer tmpBolt.Close() - // commit bolt data - err = tx.Commit() + _, err = io.Copy(w, tmpBolt) if err != nil { - return fmt.Errorf("error commit tx to backup root bolt: %v", err) + w.Close() + return fmt.Errorf("failed to copy bolt to destination: %w", err) } - return copyBolt.Sync() + // Sync and close the destination writer + if err := w.Sync(); err != nil { + w.Close() + return fmt.Errorf("failed to sync destination bolt: %w", err) + } + + if err := w.Close(); err != nil { + return fmt.Errorf("failed to close destination bolt: %w", err) + } + + return nil } func (is *IndexSnapshot) UpdateIOStats(val uint64) { diff --git a/index/scorch/vfs/fs_directory.go b/index/scorch/vfs/fs_directory.go index 531cc21d4..d8d2f69df 100644 --- a/index/scorch/vfs/fs_directory.go +++ b/index/scorch/vfs/fs_directory.go @@ -71,7 +71,7 @@ func (d *FSDirectory) Create(name string) (apivfs.WriteCloser, error) { return nil, fmt.Errorf("failed to create parent directory: %w", err) } - f, err := os.Create(fullPath) + f, err := os.OpenFile(fullPath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0644) if err != nil { return nil, err } @@ -141,6 +141,21 @@ func (d *FSDirectory) Sync() error { return nil } +// GetWriter implements index.Directory for backwards compatibility. +// This allows FSDirectory to be used where index.Directory is expected. +func (d *FSDirectory) GetWriter(filePath string) (io.WriteCloser, error) { + // Create any necessary parent directories + dir := filepath.Dir(filePath) + if dir != "" && dir != "." { + if err := d.MkdirAll(dir, 0755); err != nil { + return nil, fmt.Errorf("failed to create parent directories: %w", err) + } + } + + // Use Create to get a WriteCloser + return d.Create(filePath) +} + // Lock acquires an exclusive lock on the directory. func (d *FSDirectory) Lock() error { d.mu.Lock() diff --git a/index_test.go b/index_test.go index 7ed27ff86..93a1eab16 100644 --- a/index_test.go +++ b/index_test.go @@ -41,6 +41,7 @@ import ( index "github.com/blevesearch/bleve_index_api" "github.com/blevesearch/bleve/v2/index/scorch" + "github.com/blevesearch/bleve/v2/index/scorch/vfs" "github.com/blevesearch/bleve/v2/index/upsidedown" ) @@ -3125,7 +3126,13 @@ func TestCopyIndex(t *testing.T) { backupIndexPath := createTmpIndexPath(t) defer cleanupTmpIndexPath(t, backupIndexPath) - err = copyableIndex.CopyTo(FileSystemDirectory(backupIndexPath)) + // Create a VFS-compatible directory for the backup + backupVFSDir, err := vfs.NewFSDirectory(backupIndexPath) + if err != nil { + t.Fatalf("error creating backup VFS directory: %v", err) + } + + err = copyableIndex.CopyTo(backupVFSDir) if err != nil { t.Fatalf("error copying the index: %v", err) } From e2ba8576d4447c7190661d305451a82798b9f802 Mon Sep 17 00:00:00 2001 From: AJ Roetker Date: Wed, 12 Nov 2025 14:34:09 -0800 Subject: [PATCH 5/5] (feat/vfs) Fix copilot nits from code review --- index/scorch/vfs/directory_test.go | 32 ++++++++++++++++++++------- index/scorch/vfs/fs_directory.go | 6 +++-- index/scorch/vfs/fs_directory_test.go | 16 ++++++++++---- 3 files changed, 40 insertions(+), 14 deletions(-) diff --git a/index/scorch/vfs/directory_test.go b/index/scorch/vfs/directory_test.go index fecc860cb..18c98129b 100644 --- a/index/scorch/vfs/directory_test.go +++ b/index/scorch/vfs/directory_test.go @@ -72,8 +72,12 @@ func directoryTestSuite(t *testing.T, dir apivfs.Directory) { if err != nil { t.Fatalf("Failed to create file: %v", err) } - w.Write(testData) - w.Close() + if _, err := w.Write(testData); err != nil { + t.Fatalf("Failed to write data: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Failed to close writer: %v", err) + } // Stat file fi, err := dir.Stat(testFile) @@ -90,7 +94,9 @@ func directoryTestSuite(t *testing.T, dir apivfs.Directory) { } // Clean up - dir.Remove(testFile) + if err := dir.Remove(testFile); err != nil { + t.Fatalf("Failed to remove file: %v", err) + } }) t.Run("Rename", func(t *testing.T) { @@ -103,8 +109,12 @@ func directoryTestSuite(t *testing.T, dir apivfs.Directory) { if err != nil { t.Fatalf("Failed to create file: %v", err) } - w.Write(testData) - w.Close() + if _, err := w.Write(testData); err != nil { + t.Fatalf("Failed to write data: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Failed to close writer: %v", err) + } // Rename if err := dir.Rename(oldName, newName); err != nil { @@ -128,7 +138,9 @@ func directoryTestSuite(t *testing.T, dir apivfs.Directory) { } // Clean up - dir.Remove(newName) + if err := dir.Remove(newName); err != nil { + t.Fatalf("Failed to remove file: %v", err) + } }) t.Run("Remove", func(t *testing.T) { @@ -139,8 +151,12 @@ func directoryTestSuite(t *testing.T, dir apivfs.Directory) { if err != nil { t.Fatalf("Failed to create file: %v", err) } - w.Write([]byte("remove test")) - w.Close() + if _, err := w.Write([]byte("remove test")); err != nil { + t.Fatalf("Failed to write data: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Failed to close writer: %v", err) + } // Remove file if err := dir.Remove(testFile); err != nil { diff --git a/index/scorch/vfs/fs_directory.go b/index/scorch/vfs/fs_directory.go index d8d2f69df..cf54f25af 100644 --- a/index/scorch/vfs/fs_directory.go +++ b/index/scorch/vfs/fs_directory.go @@ -29,7 +29,7 @@ import ( type FSDirectory struct { basePath string lockFile *os.File - mu sync.RWMutex + mu sync.Mutex } // NewFSDirectory creates a new filesystem-based Directory at the given path. @@ -181,7 +181,9 @@ func (d *FSDirectory) Lock() error { // Uses platform-specific implementation (flock on Unix, LockFileEx on Windows) err = flock(f, true) if err != nil { - f.Close() + if closeErr := f.Close(); closeErr != nil { + return fmt.Errorf("failed to acquire lock (another process may have the index open): %w, and failed to close lock file: %v", err, closeErr) + } return fmt.Errorf("failed to acquire lock (another process may have the index open): %w", err) } diff --git a/index/scorch/vfs/fs_directory_test.go b/index/scorch/vfs/fs_directory_test.go index 9083db12c..79d55f77c 100644 --- a/index/scorch/vfs/fs_directory_test.go +++ b/index/scorch/vfs/fs_directory_test.go @@ -132,8 +132,12 @@ func TestFSDirectory_DirectoryOperations(t *testing.T) { if err != nil { t.Fatalf("Failed to create file %s: %v", name, err) } - w.Write([]byte("test")) - w.Close() + if _, err := w.Write([]byte("test")); err != nil { + t.Fatalf("Failed to write to file %s: %v", name, err) + } + if err := w.Close(); err != nil { + t.Fatalf("Failed to close file %s: %v", name, err) + } } // Test ReadDir @@ -216,8 +220,12 @@ func TestFSDirectory_ConcurrentReads(t *testing.T) { if err != nil { t.Fatalf("Failed to create file: %v", err) } - w.Write(testData) - w.Close() + if _, err := w.Write(testData); err != nil { + t.Fatalf("Failed to write to file: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("Failed to close file: %v", err) + } // Perform concurrent reads const numReaders = 10