10 Commits

Author SHA1 Message Date
058aff4019 Добавлена возможность асинхронного создания снапшотов
А также обновлены зависимости
2025-11-21 14:46:16 +03:00
6845219e94 Попытка исправления загрузки локального блоба 2025-10-30 02:10:01 +03:00
99764eb91f Исправление "двоения" ErrNotFound x2 2025-10-30 01:54:30 +03:00
644a94656a Исправление "двоения" ErrNotFound 2025-10-30 01:50:32 +03:00
cd98d1f4a2 Добавлена возможность зарегистрировать локальный снапшот 2025-10-30 01:36:21 +03:00
f34539c06b Добавлена локальная загрузка снапшота 2025-10-30 00:20:28 +03:00
5192658607 Refactor snapshot handling: improve directory creation, parent snapshot cleanup, and diff application logic. Update integration test to cover full and incremental update cycles. 2025-07-14 00:57:42 +03:00
7b670947c3 Add GetCurrentSnapshotID method to retrieve the current snapshot ID 2025-07-13 23:19:41 +03:00
aaf227f25a Remove obsolete gRPC client/server implementations and migrate to remote package 2025-07-13 23:12:39 +03:00
efa2bec38b Add RegisterLocalSnapshot and GetRemoteSnapshotMetadata methods
Introduce methods to register local snapshots from archives and to download or restore snapshot metadata, improving snapshot management capabilities. Update README with usage examples.
2025-07-10 12:49:05 +03:00
24 changed files with 1444 additions and 2404 deletions

View File

@@ -41,10 +41,6 @@ test-coverage:
go test -v -coverprofile=coverage.out ./...
go tool cover -html=coverage.out -o coverage.html
# Run linter
lint:
golangci-lint run
# Run all checks (tests + linter)
check: test lint

View File

@@ -223,6 +223,41 @@ func main() {
## Advanced Usage
### Registering a Local Snapshot
You can register a local snapshot from an existing archive file with a specified UUID:
```go
// Register a local snapshot from an archive file
archivePath := "/path/to/your/archive.zip"
snapshotID := "custom-uuid-for-snapshot"
snapshotName := "My Local Snapshot"
if err := ag.RegisterLocalSnapshot(ctx, archivePath, snapshotID, snapshotName); err != nil {
log.Fatalf("Failed to register local snapshot: %v", err)
}
```
### Downloading Only Snapshot Metadata
You can download only the metadata of a snapshot from a remote server without downloading the actual files:
```go
// Download only the metadata of a snapshot from a remote server
remoteAddress := "remote-server:50051"
snapshotID := "snapshot-id-to-download"
if err := ag.GetRemoteSnapshotMetadata(ctx, remoteAddress, snapshotID); err != nil {
log.Fatalf("Failed to download snapshot metadata: %v", err)
}
// If you have a local blob but missing metadata, you can restore the metadata
// by passing an empty address
if err := ag.GetRemoteSnapshotMetadata(ctx, "", snapshotID); err != nil {
log.Fatalf("Failed to restore snapshot metadata: %v", err)
}
```
### Creating Incremental Snapshots
You can create incremental snapshots by specifying a parent snapshot ID:
@@ -294,6 +329,8 @@ The main entry point for the library.
- `ConnectRemote(address string) (*grpc.SnapshotClient, error)` - Connect to a remote server
- `GetRemoteSnapshotList(ctx context.Context, address string) ([]store.SnapshotInfo, error)` - List snapshots from a remote server
- `GetRemoteSnapshot(ctx context.Context, address string, snapshotID string, localParentID string) error` - Download a snapshot from a remote server
- `RegisterLocalSnapshot(ctx context.Context, archivePath string, snapshotID string, name string) error` - Register a local snapshot from an archive path with a specified UUID
- `GetRemoteSnapshotMetadata(ctx context.Context, address string, snapshotID string) error` - Download only the metadata of a snapshot from a remote server
### AgateOptions
@@ -304,7 +341,3 @@ Configuration options for the Agate library.
- `CloseFunc func() error` - Called before a snapshot is created or restored
- `MetadataStore store.MetadataStore` - Implementation of the metadata store
- `BlobStore store.BlobStore` - Implementation of the blob store
## License
[Add your license information here]

401
api.go
View File

@@ -4,14 +4,17 @@ import (
"context"
"errors"
"fmt"
"gitea.unprism.ru/KRBL/Agate/archive"
"gitea.unprism.ru/KRBL/Agate/grpc"
"gitea.unprism.ru/KRBL/Agate/interfaces"
"io"
"log"
"os"
"path/filepath"
"sync"
"time"
"gitea.unprism.ru/KRBL/Agate/archive"
"gitea.unprism.ru/KRBL/Agate/interfaces"
"gitea.unprism.ru/KRBL/Agate/models"
"gitea.unprism.ru/KRBL/Agate/remote"
"gitea.unprism.ru/KRBL/Agate/store"
"gitea.unprism.ru/KRBL/Agate/stores"
@@ -224,6 +227,56 @@ func (a *Agate) SaveSnapshot(ctx context.Context, name string, parentID string)
return snapshot.ID, nil
}
// SnapshotAsync creates a new snapshot asynchronously.
// It returns the job ID (which is also the snapshot ID) immediately.
// The actual snapshot creation happens in a background goroutine.
// Use GetSnapshotStatus to check the progress.
func (a *Agate) SnapshotAsync(ctx context.Context, name string, parentID string) (string, error) {
a.options.Logger.Printf("Starting async snapshot creation with name: %s", name)
// If parentID is not provided, use the current snapshot ID
if parentID == "" {
parentID = a.currentSnapshotID
}
return a.manager.CreateSnapshotAsync(ctx, a.options.BlobStore.GetActiveDir(), name, parentID,
func() {
// onStart: Lock mutex and close resources
a.mutex.Lock()
if a.options.CloseFunc != nil {
if err := a.options.CloseFunc(); err != nil {
a.options.Logger.Printf("ERROR: failed to close resources before async snapshot: %v", err)
}
}
},
func(id string, err error) {
// onFinish: Open resources, update state, and unlock mutex
defer a.mutex.Unlock()
if a.options.OpenFunc != nil {
if err := a.options.OpenFunc(a.options.BlobStore.GetActiveDir()); err != nil {
a.options.Logger.Printf("ERROR: failed to open resources after async snapshot: %v", err)
}
}
if err == nil {
a.currentSnapshotID = id
if err := a.saveCurrentSnapshotID(); err != nil {
a.options.Logger.Printf("ERROR: failed to save current snapshot ID: %v", err)
}
a.options.Logger.Printf("Async snapshot %s created successfully", id)
} else {
a.options.Logger.Printf("Async snapshot creation failed: %v", err)
}
},
)
}
// GetSnapshotStatus returns the status of an asynchronous snapshot creation job.
func (a *Agate) GetSnapshotStatus(ctx context.Context, jobID string) (*store.SnapshotStatus, error) {
return a.manager.GetSnapshotStatus(ctx, jobID)
}
// RestoreSnapshot extracts a snapshot to the active directory.
func (a *Agate) RestoreSnapshot(ctx context.Context, snapshotID string) error {
a.mutex.Lock()
@@ -325,7 +378,9 @@ func (a *Agate) saveCurrentSnapshotID() error {
if a.currentSnapshotID == "" {
// If there's no current snapshot ID, remove the file if it exists
if _, err := os.Stat(a.currentIDFile); err == nil {
return os.Remove(a.currentIDFile)
if err := os.Remove(a.currentIDFile); err != nil {
return err
}
}
return nil
}
@@ -348,197 +403,201 @@ func (a *Agate) Close() error {
// StartServer starts a gRPC server to share snapshots.
func (a *Agate) StartServer(ctx context.Context, address string) error {
_, err := grpc.RunServer(ctx, a.manager, address)
if err != nil {
return fmt.Errorf("failed to start server: %w", err)
server := remote.NewServer(a.manager)
return server.Start(ctx, address)
}
// We don't store the server reference because we don't have a way to stop it yet
// In a future version, we could add a StopServer method
return nil
}
// ConnectRemote connects to a remote snapshot server.
// Returns a client that can be used to interact with the remote server.
func (a *Agate) ConnectRemote(address string) (*grpc.SnapshotClient, error) {
client, err := grpc.ConnectToServer(address)
if err != nil {
return nil, fmt.Errorf("failed to connect to remote server: %w", err)
}
return client, nil
}
// GetRemoteSnapshotList retrieves a list of snapshots from a remote server.
func (a *Agate) GetRemoteSnapshotList(ctx context.Context, address string) ([]store.SnapshotInfo, error) {
client, err := a.ConnectRemote(address)
if err != nil {
return nil, err
}
defer client.Close()
return client.ListSnapshots(ctx)
}
// GetRemoteSnapshot downloads a snapshot from a remote server.
// If localParentID is provided, it will be used to optimize the download by skipping files that already exist locally.
// GetRemoteSnapshot downloads a snapshot from a remote server, using an efficient differential update.
func (a *Agate) GetRemoteSnapshot(ctx context.Context, address string, snapshotID string, localParentID string) error {
client, err := a.ConnectRemote(address)
client, err := remote.NewClient(address)
if err != nil {
return err
}
defer client.Close()
defer func() {
if err := client.Close(); err != nil {
a.options.Logger.Printf("ERROR: failed to close client: %v", err)
}
}()
// Get the remote snapshot details
remoteSnapshot, err := client.FetchSnapshotDetails(ctx, snapshotID)
if err != nil {
return fmt.Errorf("failed to get snapshot details: %w", err)
}
// Create a temporary directory for downloading files
tempDownloadDir := filepath.Join(a.options.WorkDir, "temp_download", snapshotID)
if err := os.MkdirAll(tempDownloadDir, 0755); err != nil {
return fmt.Errorf("failed to create temporary download directory: %w", err)
}
defer os.RemoveAll(tempDownloadDir) // Clean up when done
a.options.Logger.Printf("Downloading snapshot %s from %s", snapshotID, address)
// If localParentID is provided, try to reuse files from the local parent snapshot
if localParentID != "" {
a.options.Logger.Printf("Using local parent snapshot %s for incremental download", localParentID)
// Get the local parent snapshot details
localParent, err := a.GetSnapshotDetails(ctx, localParentID)
if err != nil {
a.options.Logger.Printf("Warning: Failed to get local parent snapshot details: %v", err)
} else {
// Extract the local parent snapshot to a temporary directory
localParentDir := filepath.Join(a.options.WorkDir, "temp_download", localParentID)
if err := os.MkdirAll(localParentDir, 0755); err != nil {
a.options.Logger.Printf("Warning: Failed to create temporary directory for local parent: %v", err)
} else {
defer os.RemoveAll(localParentDir) // Clean up when done
if err := a.manager.ExtractSnapshot(ctx, localParentID, localParentDir, true); err != nil {
a.options.Logger.Printf("Warning: Failed to extract local parent snapshot: %v", err)
} else {
// Copy unchanged files from the local parent to the download directory
for _, file := range remoteSnapshot.Files {
// Skip directories, they'll be created as needed
if file.IsDir {
continue
}
// Check if the file exists in the local parent with the same hash
var localFile *store.FileInfo
for _, lf := range localParent.Files {
if lf.Path == file.Path && lf.SHA256 == file.SHA256 {
localFile = &lf
break
}
}
if localFile != nil {
// File exists in local parent with the same hash, copy it
srcPath := filepath.Join(localParentDir, localFile.Path)
dstPath := filepath.Join(tempDownloadDir, file.Path)
// Ensure the destination directory exists
if err := os.MkdirAll(filepath.Dir(dstPath), 0755); err != nil {
a.options.Logger.Printf("Failed to create directory for %s: %v", dstPath, err)
continue
}
// Copy the file
srcFile, err := os.Open(srcPath)
if err != nil {
a.options.Logger.Printf("Failed to copy file %s, will download instead: %v", file.Path, err)
continue
}
defer srcFile.Close()
dstFile, err := os.Create(dstPath)
if err != nil {
a.options.Logger.Printf("Failed to create destination file %s: %v", dstPath, err)
continue
}
_, err = io.Copy(dstFile, srcFile)
dstFile.Close()
if err != nil {
a.options.Logger.Printf("Failed to copy file data for %s: %v", file.Path, err)
// If copy fails, the file will be downloaded
} else {
a.options.Logger.Printf("Reusing file %s from local parent", file.Path)
}
}
}
}
}
}
}
// Download the snapshot files
a.options.Logger.Printf("Downloading files for snapshot %s", snapshotID)
// Get snapshot details to know what files we need to download
remoteDetails, err := client.FetchSnapshotDetails(ctx, snapshotID)
if err != nil {
return fmt.Errorf("failed to get remote snapshot details: %w", err)
}
// Check which files we already have and which we need to download
for _, file := range remoteDetails.Files {
if file.IsDir {
continue // Skip directories
tempDownloadDir := filepath.Join(a.options.WorkDir, "temp_download")
if err := os.MkdirAll(tempDownloadDir, 0755); err != nil {
return fmt.Errorf("failed to create temp download dir: %w", err)
}
filePath := filepath.Join(tempDownloadDir, file.Path)
if _, err := os.Stat(filePath); os.IsNotExist(err) {
// File doesn't exist yet, we'll need to download it
a.options.Logger.Printf("Downloading file %s", file.Path)
newSnapshotDir := filepath.Join(tempDownloadDir, "new_content_"+snapshotID)
if err := os.MkdirAll(newSnapshotDir, 0755); err != nil {
return fmt.Errorf("failed to create new snapshot directory: %w", err)
}
defer func() {
if err := os.RemoveAll(newSnapshotDir); err != nil {
a.options.Logger.Printf("ERROR: failed to remove temp dir: %v", err)
}
}()
if err := client.DownloadSnapshot(ctx, snapshotID, tempDownloadDir, localParentID); err != nil {
return fmt.Errorf("failed to download snapshot: %w", err)
}
a.options.Logger.Printf("Creating archive from downloaded files")
// Create a zip archive from the downloaded files
zipPath := filepath.Join(a.options.WorkDir, "temp_download", snapshotID+".zip")
if err := archive.CreateArchive(tempDownloadDir, zipPath); err != nil {
return fmt.Errorf("failed to create zip archive: %w", err)
}
// Store the blob with the remote snapshot ID
zipFile, err := os.Open(zipPath)
if localParentID != "" {
if err := a.manager.ExtractSnapshot(ctx, localParentID, newSnapshotDir, false); err != nil {
a.options.Logger.Printf("Warning: failed to extract local parent snapshot %s: %v", localParentID, err)
} else {
localParentSnap, err := a.GetSnapshotDetails(ctx, localParentID)
if err != nil {
return fmt.Errorf("failed to open zip file: %w", err)
}
defer zipFile.Close()
defer os.Remove(zipPath) // Clean up the zip file when done
a.options.Logger.Printf("Storing blob with ID %s", remoteSnapshot.ID)
// Store the blob with the remote snapshot ID
_, err = a.options.BlobStore.StoreBlob(ctx, remoteSnapshot.ID, zipFile)
if err != nil {
return fmt.Errorf("failed to store blob: %w", err)
a.options.Logger.Printf("Warning: failed to get local parent details %s: %v", localParentID, err)
} else {
remoteFilesMap := make(map[string]struct{})
for _, f := range remoteSnapshot.Files {
remoteFilesMap[f.Path] = struct{}{}
}
a.options.Logger.Printf("Saving snapshot metadata")
for _, localFile := range localParentSnap.Files {
if _, exists := remoteFilesMap[localFile.Path]; !exists {
pathToDelete := filepath.Join(newSnapshotDir, localFile.Path)
if err := os.RemoveAll(pathToDelete); err != nil {
a.options.Logger.Printf("Warning: failed to delete file %s during diff apply: %v", pathToDelete, err)
}
}
}
}
}
}
// Save the remote snapshot metadata
err = a.options.MetadataStore.SaveSnapshotMetadata(ctx, *remoteSnapshot)
diffArchivePath := filepath.Join(tempDownloadDir, snapshotID+"_diff.zip")
diffPartPath := diffArchivePath + ".part"
a.options.Logger.Printf("Downloading diff for snapshot %s from parent %s", snapshotID, localParentID)
if err := client.DownloadSnapshotDiff(ctx, snapshotID, localParentID, diffPartPath); err != nil {
return fmt.Errorf("failed to download snapshot diff: %w", err)
}
if err := os.Rename(diffPartPath, diffArchivePath); err != nil {
return fmt.Errorf("failed to finalize downloaded diff: %w", err)
}
defer func() {
if err := os.Remove(diffArchivePath); err != nil {
a.options.Logger.Printf("ERROR: failed to remove temp file: %v", err)
}
}()
if err := extractArchive(diffArchivePath, newSnapshotDir); err != nil {
return fmt.Errorf("failed to extract diff archive: %w", err)
}
finalArchivePath := filepath.Join(tempDownloadDir, snapshotID+".zip")
if err := archive.CreateArchive(newSnapshotDir, finalArchivePath); err != nil {
return fmt.Errorf("failed to create final snapshot archive: %w", err)
}
defer func() {
if err := os.Remove(finalArchivePath); err != nil {
a.options.Logger.Printf("ERROR: failed to remove temp file: %v", err)
}
}()
finalArchiveFile, err := os.Open(finalArchivePath)
if err != nil {
return fmt.Errorf("failed to open final archive: %w", err)
}
defer finalArchiveFile.Close()
if _, err := a.options.BlobStore.StoreBlob(ctx, snapshotID, finalArchiveFile); err != nil {
return fmt.Errorf("failed to store final blob: %w", err)
}
if err := a.options.MetadataStore.SaveSnapshotMetadata(ctx, *remoteSnapshot); err != nil {
a.options.BlobStore.DeleteBlob(ctx, snapshotID)
return fmt.Errorf("failed to save snapshot metadata: %w", err)
}
a.options.Logger.Printf("Successfully imported remote snapshot %s", snapshotID)
return nil
}
func (a *Agate) GetCurrentSnapshotID() string {
return a.currentSnapshotID
}
// RegisterLocalSnapshot регистрирует локальный файл как блоб снимка и создает
// соответствующую запись в метаданных. Если снимок с таким ID уже существует,
// метод ничего не делает и возвращает nil.
//
// - ctx: Контекст для выполнения операции.
// - snapshotID: ID регистрируемого снимка.
// - parentID: ID родительского снимка. Может быть пустым для полных снимков.
// - name: Описательное имя для снимка.
// - localPath: Абсолютный путь к локальному файлу снимка (полному или дифф-архиву).
func (ag *Agate) RegisterLocalSnapshot(ctx context.Context, snapshotID, parentID, name, localPath string) error {
// 1. Check if snapshot already exists
_, err := ag.manager.GetSnapshotDetails(ctx, snapshotID)
if err == nil {
ag.options.Logger.Printf("snapshot %s already exists, skipping registration", snapshotID)
return nil // Snapshot already exists
}
// We expect ErrNotFound, anything else is a real error.
if !errors.Is(err, models.ErrNotFound) {
return fmt.Errorf("failed to check for existing snapshot: %w", err)
}
// 2. Add the file to the blob store
// Check if blob already exists. If so, we assume it's the correct one and skip overwriting.
// This is to prevent issues when registering a file that is already in the blob store.
_, err = ag.options.BlobStore.GetBlobPath(ctx, snapshotID)
if err == nil {
ag.options.Logger.Printf("blob for snapshot %s already exists, skipping storing it", snapshotID)
} else if errors.Is(err, models.ErrNotFound) {
// Blob does not exist, so we store it.
localFile, err := os.Open(localPath)
if err != nil {
return fmt.Errorf("failed to open local snapshot file: %w", err)
}
defer func() {
if err := localFile.Close(); err != nil {
ag.options.Logger.Printf("ERROR: failed to close local file: %v", err)
}
}()
if _, err = ag.options.BlobStore.StoreBlob(ctx, snapshotID, localFile); err != nil {
return fmt.Errorf("failed to store blob from local file: %w", err)
}
} else {
// Another error occurred when checking for the blob.
return fmt.Errorf("failed to check for existing blob: %w", err)
}
// 3. Create and save snapshot metadata
// We get the file list from the archive to create the metadata.
// Note: This method does not calculate file hashes, so the metadata will be incomplete.
// This is a limitation of the current implementation.
var files []store.FileInfo
archiveFiles, err := archive.ListArchiveContents(localPath)
if err != nil {
// If we can't list the contents, we can't create the metadata.
// We should clean up the blob we just stored.
_ = ag.options.BlobStore.DeleteBlob(ctx, snapshotID)
return fmt.Errorf("failed to list archive contents for metadata creation: %w", err)
}
for _, f := range archiveFiles {
files = append(files, store.FileInfo{
Path: f.Path,
Size: int64(f.Size),
IsDir: f.IsDir,
// SHA256 is intentionally left empty as we don't have it.
})
}
snapshot := store.Snapshot{
ID: snapshotID,
Name: name,
ParentID: parentID,
CreationTime: time.Now(),
Files: files,
}
if err := ag.options.MetadataStore.SaveSnapshotMetadata(ctx, snapshot); err != nil {
// Clean up the blob
_ = ag.options.BlobStore.DeleteBlob(ctx, snapshotID)
return fmt.Errorf("failed to save snapshot metadata: %w", err)
}
ag.options.Logger.Printf("Successfully registered local snapshot %s", snapshotID)
return nil
}

View File

@@ -95,6 +95,104 @@ func CreateArchive(sourceDir, targetPath string) error {
return nil
}
// CreateArchiveWithProgress creates a ZIP archive with progress reporting.
// onProgress is called with the current number of bytes written and the total size.
func CreateArchiveWithProgress(sourceDir, targetPath string, onProgress func(current, total int64)) error {
info, err := os.Stat(sourceDir)
if err != nil {
return fmt.Errorf("failed to stat source directory %s: %w", sourceDir, err)
}
if !info.IsDir() {
return fmt.Errorf("source %s is not a directory", sourceDir)
}
// Calculate total size
var totalSize int64
err = filepath.Walk(sourceDir, func(path string, info os.FileInfo, err error) error {
if err != nil {
return err
}
if !info.IsDir() {
totalSize += info.Size()
}
return nil
})
if err != nil {
return fmt.Errorf("failed to calculate total size of %s: %w", sourceDir, err)
}
// Create file for ZIP archive
outFile, err := os.Create(targetPath)
if err != nil {
return fmt.Errorf("failed to create target archive file %s: %w", targetPath, err)
}
defer outFile.Close()
// Create zip.Writer
zipWriter := zip.NewWriter(outFile)
defer zipWriter.Close()
var currentSize int64
// Recursively walk sourceDir
err = filepath.Walk(sourceDir, func(filePath string, fileInfo os.FileInfo, walkErr error) error {
if walkErr != nil {
return fmt.Errorf("error walking path %s: %w", filePath, walkErr)
}
// Skip sourceDir itself
if filePath == sourceDir {
return nil
}
// Create relative path
relativePath := strings.TrimPrefix(filePath, sourceDir+string(filepath.Separator))
relativePath = filepath.ToSlash(relativePath)
// Check if directory
if fileInfo.IsDir() {
_, err = zipWriter.Create(relativePath + "/")
if err != nil {
return fmt.Errorf("failed to create directory entry %s in archive: %w", relativePath, err)
}
return nil
}
// Open file for reading
fileToArchive, err := os.Open(filePath)
if err != nil {
return fmt.Errorf("failed to open file %s for archiving: %w", filePath, err)
}
defer fileToArchive.Close()
// Create archive entry
zipEntryWriter, err := zipWriter.Create(relativePath)
if err != nil {
return fmt.Errorf("failed to create entry %s in archive: %w", relativePath, err)
}
// Copy content
n, err := io.Copy(zipEntryWriter, fileToArchive)
if err != nil {
return fmt.Errorf("failed to copy file content %s to archive: %w", filePath, err)
}
currentSize += n
if onProgress != nil {
onProgress(currentSize, totalSize)
}
return nil
})
if err != nil {
os.Remove(targetPath)
return fmt.Errorf("failed during directory walk for archiving %s: %w", sourceDir, err)
}
return nil
}
// ListArchiveContents читает ZIP-архив и возвращает информацию о его содержимом.
func ListArchiveContents(archivePath string) ([]ArchiveEntryInfo, error) {
// Открываем ZIP-архив

115
async_test.go Normal file
View File

@@ -0,0 +1,115 @@
package agate
import (
"context"
"os"
"path/filepath"
"testing"
"time"
)
func TestSnapshotAsync(t *testing.T) {
// Setup temporary work directory
workDir, err := os.MkdirTemp("", "agate_async_test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(workDir)
// Initialize Agate
opts := AgateOptions{
WorkDir: workDir,
Logger: nil, // Disable logging for test
}
ag, err := New(opts)
if err != nil {
t.Fatalf("Failed to initialize Agate: %v", err)
}
defer ag.Close()
// Get active directory and create some dummy files
activeDir := ag.GetActiveDir()
if err := os.MkdirAll(activeDir, 0755); err != nil {
t.Fatalf("Failed to create active dir: %v", err)
}
// Create a large-ish file to ensure it takes some time (though still fast)
dummyFile := filepath.Join(activeDir, "data.bin")
data := make([]byte, 1024*1024) // 1MB
if err := os.WriteFile(dummyFile, data, 0644); err != nil {
t.Fatalf("Failed to create dummy file: %v", err)
}
// Start async snapshot
ctx := context.Background()
snapshotID, err := ag.SnapshotAsync(ctx, "async-snap", "")
if err != nil {
t.Fatalf("SnapshotAsync failed: %v", err)
}
if snapshotID == "" {
t.Fatal("SnapshotAsync returned empty ID")
}
// Check status immediately
status, err := ag.GetSnapshotStatus(ctx, snapshotID)
if err != nil {
t.Fatalf("GetSnapshotStatus failed: %v", err)
}
// Status should be pending or running
if status.Status != "pending" && status.Status != "running" {
t.Errorf("Initial status should be pending or running, got: %s", status.Status)
}
// Poll for completion
timeout := time.After(5 * time.Second)
ticker := time.NewTicker(10 * time.Millisecond)
defer ticker.Stop()
done := false
for !done {
select {
case <-timeout:
t.Fatal("Timeout waiting for snapshot completion")
case <-ticker.C:
status, err := ag.GetSnapshotStatus(ctx, snapshotID)
if err != nil {
t.Fatalf("GetSnapshotStatus failed during polling: %v", err)
}
if status.Status == "done" {
done = true
if status.Progress != 1.0 {
t.Errorf("Expected progress 1.0, got %f", status.Progress)
}
} else if status.Status == "failed" {
t.Fatalf("Snapshot creation failed: %s", status.Error)
}
}
}
// Verify snapshot exists
snaps, err := ag.ListSnapshots(ctx)
if err != nil {
t.Fatalf("ListSnapshots failed: %v", err)
}
found := false
for _, s := range snaps {
if s.ID == snapshotID {
found = true
break
}
}
if !found {
t.Errorf("Snapshot %s not found in list", snapshotID)
}
// Verify current snapshot ID is updated
if ag.GetCurrentSnapshotID() != snapshotID {
t.Errorf("Current snapshot ID not updated. Expected %s, got %s", snapshotID, ag.GetCurrentSnapshotID())
}
}

215
go.mod
View File

@@ -2,215 +2,18 @@ module gitea.unprism.ru/KRBL/Agate
go 1.24.3
tool github.com/golangci/golangci-lint/v2/cmd/golangci-lint
require (
github.com/google/uuid v1.6.0
github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3
github.com/mattn/go-sqlite3 v1.14.28
google.golang.org/genproto/googleapis/api v0.0.0-20250505200425-f936aa4a68b2
google.golang.org/grpc v1.72.0
google.golang.org/protobuf v1.36.6
github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3
github.com/mattn/go-sqlite3 v1.14.32
google.golang.org/genproto/googleapis/api v0.0.0-20251111163417-95abcf5c77ba
google.golang.org/grpc v1.77.0
google.golang.org/protobuf v1.36.10
)
require (
4d63.com/gocheckcompilerdirectives v1.3.0 // indirect
4d63.com/gochecknoglobals v0.2.2 // indirect
github.com/4meepo/tagalign v1.4.2 // indirect
github.com/Abirdcfly/dupword v0.1.3 // indirect
github.com/Antonboom/errname v1.1.0 // indirect
github.com/Antonboom/nilnil v1.1.0 // indirect
github.com/Antonboom/testifylint v1.6.1 // indirect
github.com/BurntSushi/toml v1.5.0 // indirect
github.com/Djarvur/go-err113 v0.0.0-20210108212216-aea10b59be24 // indirect
github.com/GaijinEntertainment/go-exhaustruct/v3 v3.3.1 // indirect
github.com/Masterminds/semver/v3 v3.3.1 // indirect
github.com/OpenPeeDeeP/depguard/v2 v2.2.1 // indirect
github.com/alecthomas/chroma/v2 v2.17.2 // indirect
github.com/alecthomas/go-check-sumtype v0.3.1 // indirect
github.com/alexkohler/nakedret/v2 v2.0.6 // indirect
github.com/alexkohler/prealloc v1.0.0 // indirect
github.com/alingse/asasalint v0.0.11 // indirect
github.com/alingse/nilnesserr v0.2.0 // indirect
github.com/ashanbrown/forbidigo v1.6.0 // indirect
github.com/ashanbrown/makezero v1.2.0 // indirect
github.com/aymanbagabas/go-osc52/v2 v2.0.1 // indirect
github.com/beorn7/perks v1.0.1 // indirect
github.com/bkielbasa/cyclop v1.2.3 // indirect
github.com/blizzy78/varnamelen v0.8.0 // indirect
github.com/bombsimon/wsl/v4 v4.7.0 // indirect
github.com/breml/bidichk v0.3.3 // indirect
github.com/breml/errchkjson v0.4.1 // indirect
github.com/butuzov/ireturn v0.4.0 // indirect
github.com/butuzov/mirror v1.3.0 // indirect
github.com/catenacyber/perfsprint v0.9.1 // indirect
github.com/ccojocar/zxcvbn-go v1.0.2 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/charithe/durationcheck v0.0.10 // indirect
github.com/charmbracelet/colorprofile v0.2.3-0.20250311203215-f60798e515dc // indirect
github.com/charmbracelet/lipgloss v1.1.0 // indirect
github.com/charmbracelet/x/ansi v0.8.0 // indirect
github.com/charmbracelet/x/cellbuf v0.0.13-0.20250311204145-2c3ea96c31dd // indirect
github.com/charmbracelet/x/term v0.2.1 // indirect
github.com/chavacava/garif v0.1.0 // indirect
github.com/ckaznocha/intrange v0.3.1 // indirect
github.com/curioswitch/go-reassign v0.3.0 // indirect
github.com/daixiang0/gci v0.13.6 // indirect
github.com/dave/dst v0.27.3 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/denis-tingaikin/go-header v0.5.0 // indirect
github.com/dlclark/regexp2 v1.11.5 // indirect
github.com/ettle/strcase v0.2.0 // indirect
github.com/fatih/color v1.18.0 // indirect
github.com/fatih/structtag v1.2.0 // indirect
github.com/firefart/nonamedreturns v1.0.6 // indirect
github.com/fsnotify/fsnotify v1.5.4 // indirect
github.com/fzipp/gocyclo v0.6.0 // indirect
github.com/ghostiam/protogetter v0.3.15 // indirect
github.com/go-critic/go-critic v0.13.0 // indirect
github.com/go-toolsmith/astcast v1.1.0 // indirect
github.com/go-toolsmith/astcopy v1.1.0 // indirect
github.com/go-toolsmith/astequal v1.2.0 // indirect
github.com/go-toolsmith/astfmt v1.1.0 // indirect
github.com/go-toolsmith/astp v1.1.0 // indirect
github.com/go-toolsmith/strparse v1.1.0 // indirect
github.com/go-toolsmith/typep v1.1.0 // indirect
github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
github.com/go-xmlfmt/xmlfmt v1.1.3 // indirect
github.com/gobwas/glob v0.2.3 // indirect
github.com/gofrs/flock v0.12.1 // indirect
github.com/golang/protobuf v1.5.4 // indirect
github.com/golangci/dupl v0.0.0-20250308024227-f665c8d69b32 // indirect
github.com/golangci/go-printf-func-name v0.1.0 // indirect
github.com/golangci/gofmt v0.0.0-20250106114630-d62b90e6713d // indirect
github.com/golangci/golangci-lint/v2 v2.1.6 // indirect
github.com/golangci/golines v0.0.0-20250217134842-442fd0091d95 // indirect
github.com/golangci/misspell v0.6.0 // indirect
github.com/golangci/plugin-module-register v0.1.1 // indirect
github.com/golangci/revgrep v0.8.0 // indirect
github.com/golangci/unconvert v0.0.0-20250410112200-a129a6e6413e // indirect
github.com/google/go-cmp v0.7.0 // indirect
github.com/gordonklaus/ineffassign v0.1.0 // indirect
github.com/gostaticanalysis/analysisutil v0.7.1 // indirect
github.com/gostaticanalysis/comment v1.5.0 // indirect
github.com/gostaticanalysis/forcetypeassert v0.2.0 // indirect
github.com/gostaticanalysis/nilerr v0.1.1 // indirect
github.com/hashicorp/go-immutable-radix/v2 v2.1.0 // indirect
github.com/hashicorp/go-version v1.7.0 // indirect
github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
github.com/hashicorp/hcl v1.0.0 // indirect
github.com/hexops/gotextdiff v1.0.3 // indirect
github.com/inconshreveable/mousetrap v1.1.0 // indirect
github.com/jgautheron/goconst v1.8.1 // indirect
github.com/jingyugao/rowserrcheck v1.1.1 // indirect
github.com/jjti/go-spancheck v0.6.4 // indirect
github.com/julz/importas v0.2.0 // indirect
github.com/karamaru-alpha/copyloopvar v1.2.1 // indirect
github.com/kisielk/errcheck v1.9.0 // indirect
github.com/kkHAIKE/contextcheck v1.1.6 // indirect
github.com/kulti/thelper v0.6.3 // indirect
github.com/kunwardeep/paralleltest v1.0.14 // indirect
github.com/lasiar/canonicalheader v1.1.2 // indirect
github.com/ldez/exptostd v0.4.3 // indirect
github.com/ldez/gomoddirectives v0.6.1 // indirect
github.com/ldez/grignotin v0.9.0 // indirect
github.com/ldez/tagliatelle v0.7.1 // indirect
github.com/ldez/usetesting v0.4.3 // indirect
github.com/leonklingele/grouper v1.1.2 // indirect
github.com/lucasb-eyer/go-colorful v1.2.0 // indirect
github.com/macabu/inamedparam v0.2.0 // indirect
github.com/magiconair/properties v1.8.6 // indirect
github.com/manuelarte/funcorder v0.2.1 // indirect
github.com/maratori/testableexamples v1.0.0 // indirect
github.com/maratori/testpackage v1.1.1 // indirect
github.com/matoous/godox v1.1.0 // indirect
github.com/mattn/go-colorable v0.1.14 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/mattn/go-runewidth v0.0.16 // indirect
github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
github.com/mgechev/revive v1.9.0 // indirect
github.com/mitchellh/go-homedir v1.1.0 // indirect
github.com/mitchellh/mapstructure v1.5.0 // indirect
github.com/moricho/tparallel v0.3.2 // indirect
github.com/muesli/termenv v0.16.0 // indirect
github.com/nakabonne/nestif v0.3.1 // indirect
github.com/nishanths/exhaustive v0.12.0 // indirect
github.com/nishanths/predeclared v0.2.2 // indirect
github.com/nunnatsa/ginkgolinter v0.19.1 // indirect
github.com/olekukonko/tablewriter v0.0.5 // indirect
github.com/pelletier/go-toml v1.9.5 // indirect
github.com/pelletier/go-toml/v2 v2.2.4 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/polyfloyd/go-errorlint v1.8.0 // indirect
github.com/prometheus/client_golang v1.12.1 // indirect
github.com/prometheus/client_model v0.2.0 // indirect
github.com/prometheus/common v0.32.1 // indirect
github.com/prometheus/procfs v0.7.3 // indirect
github.com/quasilyte/go-ruleguard v0.4.4 // indirect
github.com/quasilyte/go-ruleguard/dsl v0.3.22 // indirect
github.com/quasilyte/gogrep v0.5.0 // indirect
github.com/quasilyte/regex/syntax v0.0.0-20210819130434-b3f0c404a727 // indirect
github.com/quasilyte/stdinfo v0.0.0-20220114132959-f7386bf02567 // indirect
github.com/raeperd/recvcheck v0.2.0 // indirect
github.com/rivo/uniseg v0.4.7 // indirect
github.com/rogpeppe/go-internal v1.14.1 // indirect
github.com/ryancurrah/gomodguard v1.4.1 // indirect
github.com/ryanrolds/sqlclosecheck v0.5.1 // indirect
github.com/sanposhiho/wastedassign/v2 v2.1.0 // indirect
github.com/santhosh-tekuri/jsonschema/v6 v6.0.1 // indirect
github.com/sashamelentyev/interfacebloat v1.1.0 // indirect
github.com/sashamelentyev/usestdlibvars v1.28.0 // indirect
github.com/securego/gosec/v2 v2.22.3 // indirect
github.com/sirupsen/logrus v1.9.3 // indirect
github.com/sivchari/containedctx v1.0.3 // indirect
github.com/sonatard/noctx v0.1.0 // indirect
github.com/sourcegraph/go-diff v0.7.0 // indirect
github.com/spf13/afero v1.14.0 // indirect
github.com/spf13/cast v1.5.0 // indirect
github.com/spf13/cobra v1.9.1 // indirect
github.com/spf13/jwalterweatherman v1.1.0 // indirect
github.com/spf13/pflag v1.0.6 // indirect
github.com/spf13/viper v1.12.0 // indirect
github.com/ssgreg/nlreturn/v2 v2.2.1 // indirect
github.com/stbenjam/no-sprintf-host-port v0.2.0 // indirect
github.com/stretchr/objx v0.5.2 // indirect
github.com/stretchr/testify v1.10.0 // indirect
github.com/subosito/gotenv v1.4.1 // indirect
github.com/tdakkota/asciicheck v0.4.1 // indirect
github.com/tetafro/godot v1.5.1 // indirect
github.com/timakin/bodyclose v0.0.0-20241222091800-1db5c5ca4d67 // indirect
github.com/timonwong/loggercheck v0.11.0 // indirect
github.com/tomarrell/wrapcheck/v2 v2.11.0 // indirect
github.com/tommy-muehle/go-mnd/v2 v2.5.1 // indirect
github.com/ultraware/funlen v0.2.0 // indirect
github.com/ultraware/whitespace v0.2.0 // indirect
github.com/uudashr/gocognit v1.2.0 // indirect
github.com/uudashr/iface v1.3.1 // indirect
github.com/xen0n/gosmopolitan v1.3.0 // indirect
github.com/xo/terminfo v0.0.0-20220910002029-abceb7e1c41e // indirect
github.com/yagipy/maintidx v1.0.0 // indirect
github.com/yeya24/promlinter v0.3.0 // indirect
github.com/ykadowak/zerologlint v0.1.5 // indirect
gitlab.com/bosi/decorder v0.4.2 // indirect
go-simpler.org/musttag v0.13.1 // indirect
go-simpler.org/sloglint v0.11.0 // indirect
go.augendre.info/fatcontext v0.8.0 // indirect
go.uber.org/atomic v1.7.0 // indirect
go.uber.org/automaxprocs v1.6.0 // indirect
go.uber.org/multierr v1.6.0 // indirect
go.uber.org/zap v1.24.0 // indirect
golang.org/x/exp/typeparams v0.0.0-20250210185358-939b2ce775ac // indirect
golang.org/x/mod v0.24.0 // indirect
golang.org/x/net v0.40.0 // indirect
golang.org/x/sync v0.14.0 // indirect
golang.org/x/sys v0.33.0 // indirect
golang.org/x/text v0.25.0 // indirect
golang.org/x/tools v0.32.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20250505200425-f936aa4a68b2 // indirect
gopkg.in/ini.v1 v1.67.0 // indirect
gopkg.in/yaml.v2 v2.4.0 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
honnef.co/go/tools v0.6.1 // indirect
mvdan.cc/gofumpt v0.8.0 // indirect
mvdan.cc/unparam v0.0.0-20250301125049-0df0534333a4 // indirect
golang.org/x/net v0.47.0 // indirect
golang.org/x/sys v0.38.0 // indirect
golang.org/x/text v0.31.0 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20251111163417-95abcf5c77ba // indirect
)

1055
go.sum

File diff suppressed because it is too large Load Diff

View File

@@ -1,236 +0,0 @@
package grpc
import (
"context"
"fmt"
"io"
"os"
"path/filepath"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"gitea.unprism.ru/KRBL/Agate/store"
)
// SnapshotClient implements the client for connecting to a remote snapshot server
type SnapshotClient struct {
conn *grpc.ClientConn
client SnapshotServiceClient
}
// NewSnapshotClient creates a new client connected to the specified address
func NewSnapshotClient(address string) (*SnapshotClient, error) {
// Connect to the server with insecure credentials (for simplicity)
conn, err := grpc.NewClient(address, grpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
return nil, fmt.Errorf("failed to connect to server at %s: %w", address, err)
}
// Create the gRPC client
client := NewSnapshotServiceClient(conn)
return &SnapshotClient{
conn: conn,
client: client,
}, nil
}
// Close closes the connection to the server
func (c *SnapshotClient) Close() error {
if c.conn != nil {
return c.conn.Close()
}
return nil
}
// ListSnapshots retrieves a list of snapshots from the remote server
func (c *SnapshotClient) ListSnapshots(ctx context.Context) ([]store.SnapshotInfo, error) {
response, err := c.client.ListSnapshots(ctx, &ListSnapshotsRequest{})
if err != nil {
return nil, fmt.Errorf("failed to list snapshots: %w", err)
}
// Convert gRPC snapshot info to store.SnapshotInfo
snapshots := make([]store.SnapshotInfo, 0, len(response.Snapshots))
for _, snapshot := range response.Snapshots {
snapshots = append(snapshots, store.SnapshotInfo{
ID: snapshot.Id,
Name: snapshot.Name,
ParentID: snapshot.ParentId,
CreationTime: snapshot.CreationTime.AsTime(),
})
}
return snapshots, nil
}
// FetchSnapshotDetails retrieves detailed information about a specific snapshot
func (c *SnapshotClient) FetchSnapshotDetails(ctx context.Context, snapshotID string) (*store.Snapshot, error) {
response, err := c.client.GetSnapshotDetails(ctx, &GetSnapshotDetailsRequest{
SnapshotId: snapshotID,
})
if err != nil {
return nil, fmt.Errorf("failed to get snapshot details: %w", err)
}
// Convert gRPC snapshot details to store.Snapshot
snapshot := &store.Snapshot{
ID: response.Info.Id,
Name: response.Info.Name,
ParentID: response.Info.ParentId,
CreationTime: response.Info.CreationTime.AsTime(),
Files: make([]store.FileInfo, 0, len(response.Files)),
}
// Convert file info
for _, file := range response.Files {
snapshot.Files = append(snapshot.Files, store.FileInfo{
Path: file.Path,
Size: file.SizeBytes,
IsDir: file.IsDir,
SHA256: file.Sha256Hash,
})
}
return snapshot, nil
}
// DownloadSnapshot downloads a snapshot from the server
// This implementation downloads each file individually to optimize bandwidth usage
func (c *SnapshotClient) DownloadSnapshot(ctx context.Context, snapshotID string, targetDir string, localParentID string) error {
// Get snapshot details
snapshot, err := c.FetchSnapshotDetails(ctx, snapshotID)
if err != nil {
return fmt.Errorf("failed to get snapshot details: %w", err)
}
// Create target directory if it doesn't exist
if err := os.MkdirAll(targetDir, 0755); err != nil {
return fmt.Errorf("failed to create target directory: %w", err)
}
// If a local parent is specified, get its details to compare files
var localParentFiles map[string]store.FileInfo
if localParentID != "" {
localParent, err := c.FetchSnapshotDetails(ctx, localParentID)
if err == nil {
// Create a map of file paths to file info for quick lookup
localParentFiles = make(map[string]store.FileInfo, len(localParent.Files))
for _, file := range localParent.Files {
localParentFiles[file.Path] = file
}
}
}
// Download each file
for _, file := range snapshot.Files {
// Skip directories, we'll create them when needed
if file.IsDir {
// Create directory
dirPath := filepath.Join(targetDir, file.Path)
if err := os.MkdirAll(dirPath, 0755); err != nil {
return fmt.Errorf("failed to create directory %s: %w", dirPath, err)
}
continue
}
// Check if we can skip downloading this file
if localParentFiles != nil {
if parentFile, exists := localParentFiles[file.Path]; exists && parentFile.SHA256 == file.SHA256 {
// File exists in parent with same hash, copy it instead of downloading
parentFilePath := filepath.Join(targetDir, "..", localParentID, file.Path)
targetFilePath := filepath.Join(targetDir, file.Path)
// Ensure parent directory exists
if err := os.MkdirAll(filepath.Dir(targetFilePath), 0755); err != nil {
return fmt.Errorf("failed to create directory for %s: %w", targetFilePath, err)
}
// Copy the file
if err := copyFile(parentFilePath, targetFilePath); err != nil {
// If copy fails, fall back to downloading
fmt.Printf("Failed to copy file %s, will download instead: %v\n", file.Path, err)
} else {
// Skip to next file
continue
}
}
}
// Download the file
if err := c.downloadFile(ctx, snapshotID, file.Path, filepath.Join(targetDir, file.Path)); err != nil {
return fmt.Errorf("failed to download file %s: %w", file.Path, err)
}
}
return nil
}
// downloadFile downloads a single file from the server
func (c *SnapshotClient) downloadFile(ctx context.Context, snapshotID, filePath, targetPath string) error {
// Create the request
req := &DownloadFileRequest{
SnapshotId: snapshotID,
FilePath: filePath,
}
// Start streaming the file
stream, err := c.client.DownloadFile(ctx, req)
if err != nil {
return fmt.Errorf("failed to start file download: %w", err)
}
// Ensure the target directory exists
if err := os.MkdirAll(filepath.Dir(targetPath), 0755); err != nil {
return fmt.Errorf("failed to create directory for %s: %w", targetPath, err)
}
// Create the target file
file, err := os.Create(targetPath)
if err != nil {
return fmt.Errorf("failed to create file %s: %w", targetPath, err)
}
defer file.Close()
// Receive and write chunks
for {
resp, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
return fmt.Errorf("error receiving file chunk: %w", err)
}
// Write the chunk to the file
if _, err := file.Write(resp.ChunkData); err != nil {
return fmt.Errorf("error writing to file: %w", err)
}
}
return nil
}
// Helper function to copy a file
func copyFile(src, dst string) error {
sourceFile, err := os.Open(src)
if err != nil {
return err
}
defer sourceFile.Close()
destFile, err := os.Create(dst)
if err != nil {
return err
}
defer destFile.Close()
_, err = io.Copy(destFile, sourceFile)
return err
}
// ConnectToServer creates a new client connected to the specified address
func ConnectToServer(address string) (*SnapshotClient, error) {
return NewSnapshotClient(address)
}

View File

@@ -1,158 +0,0 @@
package grpc
import (
"context"
"fmt"
"io"
"net"
"google.golang.org/grpc"
"google.golang.org/protobuf/types/known/timestamppb"
"gitea.unprism.ru/KRBL/Agate/interfaces"
"gitea.unprism.ru/KRBL/Agate/store"
)
// SnapshotServer implements the gRPC server for snapshots
type SnapshotServer struct {
UnimplementedSnapshotServiceServer
manager interfaces.SnapshotManager
server *grpc.Server
}
// NewSnapshotServer creates a new snapshot server
func NewSnapshotServer(manager interfaces.SnapshotManager) *SnapshotServer {
return &SnapshotServer{
manager: manager,
}
}
// Start starts the gRPC server on the specified address
func (s *SnapshotServer) Start(ctx context.Context, address string) error {
lis, err := net.Listen("tcp", address)
if err != nil {
return fmt.Errorf("failed to listen on %s: %w", address, err)
}
s.server = grpc.NewServer()
RegisterSnapshotServiceServer(s.server, s)
go func() {
if err := s.server.Serve(lis); err != nil {
fmt.Printf("Server error: %v\n", err)
}
}()
fmt.Printf("Server started on %s\n", address)
return nil
}
// Stop gracefully stops the server
func (s *SnapshotServer) Stop(ctx context.Context) error {
if s.server != nil {
s.server.GracefulStop()
fmt.Println("Server stopped")
}
return nil
}
// ListSnapshots implements the gRPC ListSnapshots method
func (s *SnapshotServer) ListSnapshots(ctx context.Context, req *ListSnapshotsRequest) (*ListSnapshotsResponse, error) {
// Create empty ListOptions since the proto doesn't have active filter/pagination fields yet
opts := store.ListOptions{}
// Call manager with the required ListOptions parameter
snapshots, err := s.manager.ListSnapshots(ctx, opts)
if err != nil {
return nil, fmt.Errorf("failed to list snapshots: %w", err)
}
response := &ListSnapshotsResponse{
Snapshots: make([]*SnapshotInfo, 0, len(snapshots)),
}
for _, snapshot := range snapshots {
response.Snapshots = append(response.Snapshots, convertToGrpcSnapshotInfo(snapshot))
}
return response, nil
}
// GetSnapshotDetails implements the gRPC GetSnapshotDetails method
func (s *SnapshotServer) GetSnapshotDetails(ctx context.Context, req *GetSnapshotDetailsRequest) (*SnapshotDetails, error) {
snapshot, err := s.manager.GetSnapshotDetails(ctx, req.SnapshotId)
if err != nil {
return nil, fmt.Errorf("failed to get snapshot details: %w", err)
}
response := &SnapshotDetails{
Info: convertToGrpcSnapshotInfo(store.SnapshotInfo{
ID: snapshot.ID,
Name: snapshot.Name,
ParentID: snapshot.ParentID,
CreationTime: snapshot.CreationTime,
}),
Files: make([]*FileInfo, 0, len(snapshot.Files)),
}
for _, file := range snapshot.Files {
response.Files = append(response.Files, &FileInfo{
Path: file.Path,
SizeBytes: file.Size,
Sha256Hash: file.SHA256,
IsDir: file.IsDir,
})
}
return response, nil
}
// DownloadFile implements the gRPC DownloadFile method
func (s *SnapshotServer) DownloadFile(req *DownloadFileRequest, stream grpc.ServerStreamingServer[DownloadFileResponse]) error {
// Open the file from the snapshot
fileReader, err := s.manager.OpenFile(context.Background(), req.SnapshotId, req.FilePath)
if err != nil {
return fmt.Errorf("failed to open file: %w", err)
}
defer fileReader.Close()
// Read the file in chunks and send them to the client
buffer := make([]byte, 64*1024) // 64KB chunks
for {
n, err := fileReader.Read(buffer)
if err == io.EOF {
break
}
if err != nil {
return fmt.Errorf("failed to read file: %w", err)
}
// Send the chunk to the client
if err := stream.Send(&DownloadFileResponse{
ChunkData: buffer[:n],
}); err != nil {
return fmt.Errorf("failed to send chunk: %w", err)
}
}
return nil
}
// Helper function to convert store.SnapshotInfo to grpc.SnapshotInfo
func convertToGrpcSnapshotInfo(info store.SnapshotInfo) *SnapshotInfo {
return &SnapshotInfo{
Id: info.ID,
Name: info.Name,
ParentId: info.ParentID,
CreationTime: timestamppb.New(info.CreationTime),
}
}
// RunServer is a helper function to create and start a snapshot server
func RunServer(ctx context.Context, manager interfaces.SnapshotManager, address string) (*SnapshotServer, error) {
server := NewSnapshotServer(manager)
if err := server.Start(ctx, address); err != nil {
return nil, err
}
return server, nil
}

View File

@@ -1,7 +1,7 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.36.6
// protoc v4.25.3
// protoc-gen-go v1.36.8
// protoc v6.32.0
// source: snapshot.proto
package grpc
@@ -26,10 +26,10 @@ const (
// Метаданные файла внутри снапшота
type FileInfo struct {
state protoimpl.MessageState `protogen:"open.v1"`
Path string `protobuf:"bytes,1,opt,name=path,proto3" json:"path,omitempty"` // Относительный путь файла внутри снапшота
SizeBytes int64 `protobuf:"varint,2,opt,name=size_bytes,json=sizeBytes,proto3" json:"size_bytes,omitempty"` // Размер файла в байтах
Sha256Hash string `protobuf:"bytes,3,opt,name=sha256_hash,json=sha256Hash,proto3" json:"sha256_hash,omitempty"` // Хеш-сумма файла (SHA256)
IsDir bool `protobuf:"varint,4,opt,name=is_dir,json=isDir,proto3" json:"is_dir,omitempty"` // Является ли запись директорией
Path string `protobuf:"bytes,1,opt,name=path,proto3" json:"path,omitempty"`
SizeBytes int64 `protobuf:"varint,2,opt,name=size_bytes,json=sizeBytes,proto3" json:"size_bytes,omitempty"`
Sha256Hash string `protobuf:"bytes,3,opt,name=sha256_hash,json=sha256Hash,proto3" json:"sha256_hash,omitempty"`
IsDir bool `protobuf:"varint,4,opt,name=is_dir,json=isDir,proto3" json:"is_dir,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -95,10 +95,10 @@ func (x *FileInfo) GetIsDir() bool {
// Краткая информация о снапшоте
type SnapshotInfo struct {
state protoimpl.MessageState `protogen:"open.v1"`
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` // Уникальный ID снапшота (UUID)
Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` // Имя снапшота
ParentId string `protobuf:"bytes,3,opt,name=parent_id,json=parentId,proto3" json:"parent_id,omitempty"` // ID родительского снапшота (может быть пустым)
CreationTime *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=creation_time,json=creationTime,proto3" json:"creation_time,omitempty"` // Время создания
Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"`
ParentId string `protobuf:"bytes,3,opt,name=parent_id,json=parentId,proto3" json:"parent_id,omitempty"`
CreationTime *timestamppb.Timestamp `protobuf:"bytes,4,opt,name=creation_time,json=creationTime,proto3" json:"creation_time,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -164,8 +164,8 @@ func (x *SnapshotInfo) GetCreationTime() *timestamppb.Timestamp {
// Детальная информация о снапшоте
type SnapshotDetails struct {
state protoimpl.MessageState `protogen:"open.v1"`
Info *SnapshotInfo `protobuf:"bytes,1,opt,name=info,proto3" json:"info,omitempty"` // Краткая информация
Files []*FileInfo `protobuf:"bytes,2,rep,name=files,proto3" json:"files,omitempty"` // Список файлов в снапшоте
Info *SnapshotInfo `protobuf:"bytes,1,opt,name=info,proto3" json:"info,omitempty"`
Files []*FileInfo `protobuf:"bytes,2,rep,name=files,proto3" json:"files,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -214,7 +214,7 @@ func (x *SnapshotDetails) GetFiles() []*FileInfo {
return nil
}
// Запрос на получение списка снапшотов (можно добавить фильтры/пагинацию)
// Запрос на получение списка снапшотов
type ListSnapshotsRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
unknownFields protoimpl.UnknownFields
@@ -254,7 +254,7 @@ func (*ListSnapshotsRequest) Descriptor() ([]byte, []int) {
// Ответ со списком снапшотов
type ListSnapshotsResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Snapshots []*SnapshotInfo `protobuf:"bytes,1,rep,name=snapshots,proto3" json:"snapshots,omitempty"` // string next_page_token = 2;
Snapshots []*SnapshotInfo `protobuf:"bytes,1,rep,name=snapshots,proto3" json:"snapshots,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -299,7 +299,7 @@ func (x *ListSnapshotsResponse) GetSnapshots() []*SnapshotInfo {
// Запрос на получение деталей снапшота
type GetSnapshotDetailsRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
SnapshotId string `protobuf:"bytes,1,opt,name=snapshot_id,json=snapshotId,proto3" json:"snapshot_id,omitempty"` // ID нужного снапшота
SnapshotId string `protobuf:"bytes,1,opt,name=snapshot_id,json=snapshotId,proto3" json:"snapshot_id,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -344,8 +344,8 @@ func (x *GetSnapshotDetailsRequest) GetSnapshotId() string {
// Запрос на скачивание файла
type DownloadFileRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
SnapshotId string `protobuf:"bytes,1,opt,name=snapshot_id,json=snapshotId,proto3" json:"snapshot_id,omitempty"` // ID снапшота
FilePath string `protobuf:"bytes,2,opt,name=file_path,json=filePath,proto3" json:"file_path,omitempty"` // Путь к файлу внутри снапшота
SnapshotId string `protobuf:"bytes,1,opt,name=snapshot_id,json=snapshotId,proto3" json:"snapshot_id,omitempty"`
FilePath string `protobuf:"bytes,2,opt,name=file_path,json=filePath,proto3" json:"file_path,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -397,7 +397,7 @@ func (x *DownloadFileRequest) GetFilePath() string {
// Ответ (часть файла) при скачивании
type DownloadFileResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
ChunkData []byte `protobuf:"bytes,1,opt,name=chunk_data,json=chunkData,proto3" json:"chunk_data,omitempty"` // Кусочек данных файла
ChunkData []byte `protobuf:"bytes,1,opt,name=chunk_data,json=chunkData,proto3" json:"chunk_data,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -439,6 +439,173 @@ func (x *DownloadFileResponse) GetChunkData() []byte {
return nil
}
// Запрос на скачивание разницы между снапшотами
type DownloadSnapshotDiffRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
SnapshotId string `protobuf:"bytes,1,opt,name=snapshot_id,json=snapshotId,proto3" json:"snapshot_id,omitempty"` // ID целевого снапшота
LocalParentId string `protobuf:"bytes,2,opt,name=local_parent_id,json=localParentId,proto3" json:"local_parent_id,omitempty"` // ID снапшота, который уже есть у клиента
Offset int64 `protobuf:"varint,3,opt,name=offset,proto3" json:"offset,omitempty"` // Смещение в байтах для докачки
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *DownloadSnapshotDiffRequest) Reset() {
*x = DownloadSnapshotDiffRequest{}
mi := &file_snapshot_proto_msgTypes[8]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *DownloadSnapshotDiffRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*DownloadSnapshotDiffRequest) ProtoMessage() {}
func (x *DownloadSnapshotDiffRequest) ProtoReflect() protoreflect.Message {
mi := &file_snapshot_proto_msgTypes[8]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use DownloadSnapshotDiffRequest.ProtoReflect.Descriptor instead.
func (*DownloadSnapshotDiffRequest) Descriptor() ([]byte, []int) {
return file_snapshot_proto_rawDescGZIP(), []int{8}
}
func (x *DownloadSnapshotDiffRequest) GetSnapshotId() string {
if x != nil {
return x.SnapshotId
}
return ""
}
func (x *DownloadSnapshotDiffRequest) GetLocalParentId() string {
if x != nil {
return x.LocalParentId
}
return ""
}
func (x *DownloadSnapshotDiffRequest) GetOffset() int64 {
if x != nil {
return x.Offset
}
return 0
}
// Запрос на получение информации о дифе
type GetDiffInfoRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
SnapshotId string `protobuf:"bytes,1,opt,name=snapshot_id,json=snapshotId,proto3" json:"snapshot_id,omitempty"`
LocalParentId string `protobuf:"bytes,2,opt,name=local_parent_id,json=localParentId,proto3" json:"local_parent_id,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *GetDiffInfoRequest) Reset() {
*x = GetDiffInfoRequest{}
mi := &file_snapshot_proto_msgTypes[9]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *GetDiffInfoRequest) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*GetDiffInfoRequest) ProtoMessage() {}
func (x *GetDiffInfoRequest) ProtoReflect() protoreflect.Message {
mi := &file_snapshot_proto_msgTypes[9]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use GetDiffInfoRequest.ProtoReflect.Descriptor instead.
func (*GetDiffInfoRequest) Descriptor() ([]byte, []int) {
return file_snapshot_proto_rawDescGZIP(), []int{9}
}
func (x *GetDiffInfoRequest) GetSnapshotId() string {
if x != nil {
return x.SnapshotId
}
return ""
}
func (x *GetDiffInfoRequest) GetLocalParentId() string {
if x != nil {
return x.LocalParentId
}
return ""
}
// Информация о дифе
type DiffInfo struct {
state protoimpl.MessageState `protogen:"open.v1"`
Sha256Hash string `protobuf:"bytes,1,opt,name=sha256_hash,json=sha256Hash,proto3" json:"sha256_hash,omitempty"`
SizeBytes int64 `protobuf:"varint,2,opt,name=size_bytes,json=sizeBytes,proto3" json:"size_bytes,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
func (x *DiffInfo) Reset() {
*x = DiffInfo{}
mi := &file_snapshot_proto_msgTypes[10]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *DiffInfo) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*DiffInfo) ProtoMessage() {}
func (x *DiffInfo) ProtoReflect() protoreflect.Message {
mi := &file_snapshot_proto_msgTypes[10]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use DiffInfo.ProtoReflect.Descriptor instead.
func (*DiffInfo) Descriptor() ([]byte, []int) {
return file_snapshot_proto_rawDescGZIP(), []int{10}
}
func (x *DiffInfo) GetSha256Hash() string {
if x != nil {
return x.Sha256Hash
}
return ""
}
func (x *DiffInfo) GetSizeBytes() int64 {
if x != nil {
return x.SizeBytes
}
return 0
}
var File_snapshot_proto protoreflect.FileDescriptor
const file_snapshot_proto_rawDesc = "" +
@@ -472,11 +639,27 @@ const file_snapshot_proto_rawDesc = "" +
"\tfile_path\x18\x02 \x01(\tR\bfilePath\"5\n" +
"\x14DownloadFileResponse\x12\x1d\n" +
"\n" +
"chunk_data\x18\x01 \x01(\fR\tchunkData2\x8a\x03\n" +
"chunk_data\x18\x01 \x01(\fR\tchunkData\"~\n" +
"\x1bDownloadSnapshotDiffRequest\x12\x1f\n" +
"\vsnapshot_id\x18\x01 \x01(\tR\n" +
"snapshotId\x12&\n" +
"\x0flocal_parent_id\x18\x02 \x01(\tR\rlocalParentId\x12\x16\n" +
"\x06offset\x18\x03 \x01(\x03R\x06offset\"]\n" +
"\x12GetDiffInfoRequest\x12\x1f\n" +
"\vsnapshot_id\x18\x01 \x01(\tR\n" +
"snapshotId\x12&\n" +
"\x0flocal_parent_id\x18\x02 \x01(\tR\rlocalParentId\"J\n" +
"\bDiffInfo\x12\x1f\n" +
"\vsha256_hash\x18\x01 \x01(\tR\n" +
"sha256Hash\x12\x1d\n" +
"\n" +
"size_bytes\x18\x02 \x01(\x03R\tsizeBytes2\xb8\x04\n" +
"\x0fSnapshotService\x12k\n" +
"\rListSnapshots\x12 .agate.grpc.ListSnapshotsRequest\x1a!.agate.grpc.ListSnapshotsResponse\"\x15\x82\xd3\xe4\x93\x02\x0f\x12\r/v1/snapshots\x12}\n" +
"\x12GetSnapshotDetails\x12%.agate.grpc.GetSnapshotDetailsRequest\x1a\x1b.agate.grpc.SnapshotDetails\"#\x82\xd3\xe4\x93\x02\x1d\x12\x1b/v1/snapshots/{snapshot_id}\x12\x8a\x01\n" +
"\fDownloadFile\x12\x1f.agate.grpc.DownloadFileRequest\x1a .agate.grpc.DownloadFileResponse\"5\x82\xd3\xe4\x93\x02/\x12-/v1/snapshots/{snapshot_id}/files/{file_path}0\x01B\"Z gitea.unprism.ru/KRBL/Agate/grpcb\x06proto3"
"\fDownloadFile\x12\x1f.agate.grpc.DownloadFileRequest\x1a .agate.grpc.DownloadFileResponse\"5\x82\xd3\xe4\x93\x02/\x12-/v1/snapshots/{snapshot_id}/files/{file_path}0\x01\x12e\n" +
"\x14DownloadSnapshotDiff\x12'.agate.grpc.DownloadSnapshotDiffRequest\x1a .agate.grpc.DownloadFileResponse\"\x000\x01\x12E\n" +
"\vGetDiffInfo\x12\x1e.agate.grpc.GetDiffInfoRequest\x1a\x14.agate.grpc.DiffInfo\"\x00B\"Z gitea.unprism.ru/KRBL/Agate/grpcb\x06proto3"
var (
file_snapshot_proto_rawDescOnce sync.Once
@@ -490,7 +673,7 @@ func file_snapshot_proto_rawDescGZIP() []byte {
return file_snapshot_proto_rawDescData
}
var file_snapshot_proto_msgTypes = make([]protoimpl.MessageInfo, 8)
var file_snapshot_proto_msgTypes = make([]protoimpl.MessageInfo, 11)
var file_snapshot_proto_goTypes = []any{
(*FileInfo)(nil), // 0: agate.grpc.FileInfo
(*SnapshotInfo)(nil), // 1: agate.grpc.SnapshotInfo
@@ -500,21 +683,28 @@ var file_snapshot_proto_goTypes = []any{
(*GetSnapshotDetailsRequest)(nil), // 5: agate.grpc.GetSnapshotDetailsRequest
(*DownloadFileRequest)(nil), // 6: agate.grpc.DownloadFileRequest
(*DownloadFileResponse)(nil), // 7: agate.grpc.DownloadFileResponse
(*timestamppb.Timestamp)(nil), // 8: google.protobuf.Timestamp
(*DownloadSnapshotDiffRequest)(nil), // 8: agate.grpc.DownloadSnapshotDiffRequest
(*GetDiffInfoRequest)(nil), // 9: agate.grpc.GetDiffInfoRequest
(*DiffInfo)(nil), // 10: agate.grpc.DiffInfo
(*timestamppb.Timestamp)(nil), // 11: google.protobuf.Timestamp
}
var file_snapshot_proto_depIdxs = []int32{
8, // 0: agate.grpc.SnapshotInfo.creation_time:type_name -> google.protobuf.Timestamp
11, // 0: agate.grpc.SnapshotInfo.creation_time:type_name -> google.protobuf.Timestamp
1, // 1: agate.grpc.SnapshotDetails.info:type_name -> agate.grpc.SnapshotInfo
0, // 2: agate.grpc.SnapshotDetails.files:type_name -> agate.grpc.FileInfo
1, // 3: agate.grpc.ListSnapshotsResponse.snapshots:type_name -> agate.grpc.SnapshotInfo
3, // 4: agate.grpc.SnapshotService.ListSnapshots:input_type -> agate.grpc.ListSnapshotsRequest
5, // 5: agate.grpc.SnapshotService.GetSnapshotDetails:input_type -> agate.grpc.GetSnapshotDetailsRequest
6, // 6: agate.grpc.SnapshotService.DownloadFile:input_type -> agate.grpc.DownloadFileRequest
4, // 7: agate.grpc.SnapshotService.ListSnapshots:output_type -> agate.grpc.ListSnapshotsResponse
2, // 8: agate.grpc.SnapshotService.GetSnapshotDetails:output_type -> agate.grpc.SnapshotDetails
7, // 9: agate.grpc.SnapshotService.DownloadFile:output_type -> agate.grpc.DownloadFileResponse
7, // [7:10] is the sub-list for method output_type
4, // [4:7] is the sub-list for method input_type
8, // 7: agate.grpc.SnapshotService.DownloadSnapshotDiff:input_type -> agate.grpc.DownloadSnapshotDiffRequest
9, // 8: agate.grpc.SnapshotService.GetDiffInfo:input_type -> agate.grpc.GetDiffInfoRequest
4, // 9: agate.grpc.SnapshotService.ListSnapshots:output_type -> agate.grpc.ListSnapshotsResponse
2, // 10: agate.grpc.SnapshotService.GetSnapshotDetails:output_type -> agate.grpc.SnapshotDetails
7, // 11: agate.grpc.SnapshotService.DownloadFile:output_type -> agate.grpc.DownloadFileResponse
7, // 12: agate.grpc.SnapshotService.DownloadSnapshotDiff:output_type -> agate.grpc.DownloadFileResponse
10, // 13: agate.grpc.SnapshotService.GetDiffInfo:output_type -> agate.grpc.DiffInfo
9, // [9:14] is the sub-list for method output_type
4, // [4:9] is the sub-list for method input_type
4, // [4:4] is the sub-list for extension type_name
4, // [4:4] is the sub-list for extension extendee
0, // [0:4] is the sub-list for field type_name
@@ -531,7 +721,7 @@ func file_snapshot_proto_init() {
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: unsafe.Slice(unsafe.StringData(file_snapshot_proto_rawDesc), len(file_snapshot_proto_rawDesc)),
NumEnums: 0,
NumMessages: 8,
NumMessages: 11,
NumExtensions: 0,
NumServices: 1,
},

View File

@@ -40,7 +40,9 @@ func request_SnapshotService_ListSnapshots_0(ctx context.Context, marshaler runt
protoReq ListSnapshotsRequest
metadata runtime.ServerMetadata
)
io.Copy(io.Discard, req.Body)
if req.Body != nil {
_, _ = io.Copy(io.Discard, req.Body)
}
msg, err := client.ListSnapshots(ctx, &protoReq, grpc.Header(&metadata.HeaderMD), grpc.Trailer(&metadata.TrailerMD))
return msg, metadata, err
}
@@ -60,7 +62,9 @@ func request_SnapshotService_GetSnapshotDetails_0(ctx context.Context, marshaler
metadata runtime.ServerMetadata
err error
)
io.Copy(io.Discard, req.Body)
if req.Body != nil {
_, _ = io.Copy(io.Discard, req.Body)
}
val, ok := pathParams["snapshot_id"]
if !ok {
return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "snapshot_id")
@@ -97,7 +101,9 @@ func request_SnapshotService_DownloadFile_0(ctx context.Context, marshaler runti
metadata runtime.ServerMetadata
err error
)
io.Copy(io.Discard, req.Body)
if req.Body != nil {
_, _ = io.Copy(io.Discard, req.Body)
}
val, ok := pathParams["snapshot_id"]
if !ok {
return nil, metadata, status.Errorf(codes.InvalidArgument, "missing parameter %s", "snapshot_id")

View File

@@ -3,7 +3,7 @@ syntax = "proto3";
package agate.grpc;
import "google/protobuf/timestamp.proto";
import "google/api/annotations.proto"; // Добавлено для HTTP mapping
import "google/api/annotations.proto";
option go_package = "gitea.unprism.ru/KRBL/Agate/grpc";
@@ -30,77 +30,74 @@ service SnapshotService {
};
}
// --- Методы для управления (опционально, можно не включать в публичный API клиента) ---
// Создать новый снапшот из директории (если серверу позволено инициировать)
// rpc CreateSnapshot(CreateSnapshotRequest) returns (Snapshot);
// Удалить снапшот (если требуется)
// rpc DeleteSnapshot(DeleteSnapshotRequest) returns (DeleteSnapshotResponse);
// Скачать архив, содержащий только разницу между двумя снапшотами
rpc DownloadSnapshotDiff(DownloadSnapshotDiffRequest) returns (stream DownloadFileResponse) {}
// Получить информацию о дифе (хеш и размер)
rpc GetDiffInfo(GetDiffInfoRequest) returns (DiffInfo) {}
}
// Метаданные файла внутри снапшота
message FileInfo {
string path = 1; // Относительный путь файла внутри снапшота
int64 size_bytes = 2; // Размер файла в байтах
string sha256_hash = 3; // Хеш-сумма файла (SHA256)
bool is_dir = 4; // Является ли запись директорией
string path = 1;
int64 size_bytes = 2;
string sha256_hash = 3;
bool is_dir = 4;
}
// Краткая информация о снапшоте
message SnapshotInfo {
string id = 1; // Уникальный ID снапшота (UUID)
string name = 2; // Имя снапшота
string parent_id = 3; // ID родительского снапшота (может быть пустым)
google.protobuf.Timestamp creation_time = 4; // Время создания
string id = 1;
string name = 2;
string parent_id = 3;
google.protobuf.Timestamp creation_time = 4;
}
// Детальная информация о снапшоте
message SnapshotDetails {
SnapshotInfo info = 1; // Краткая информация
repeated FileInfo files = 2; // Список файлов в снапшоте
SnapshotInfo info = 1;
repeated FileInfo files = 2;
}
// Запрос на получение списка снапшотов (можно добавить фильтры/пагинацию)
message ListSnapshotsRequest {
// string filter_by_name = 1;
// int32 page_size = 2;
// string page_token = 3;
}
// Запрос на получение списка снапшотов
message ListSnapshotsRequest {}
// Ответ со списком снапшотов
message ListSnapshotsResponse {
repeated SnapshotInfo snapshots = 1;
// string next_page_token = 2;
}
// Запрос на получение деталей снапшота
message GetSnapshotDetailsRequest {
string snapshot_id = 1; // ID нужного снапшота
string snapshot_id = 1;
}
// Запрос на скачивание файла
message DownloadFileRequest {
string snapshot_id = 1; // ID снапшота
string file_path = 2; // Путь к файлу внутри снапшота
string snapshot_id = 1;
string file_path = 2;
}
// Ответ (часть файла) при скачивании
message DownloadFileResponse {
bytes chunk_data = 1; // Кусочек данных файла
bytes chunk_data = 1;
}
// --- Сообщения для опциональных методов управления ---
/*
message CreateSnapshotRequest {
string source_path = 1; // Путь к директории на сервере
string name = 2;
string parent_id = 3; // Опционально
// Запрос на скачивание разницы между снапшотами
message DownloadSnapshotDiffRequest {
string snapshot_id = 1; // ID целевого снапшота
string local_parent_id = 2; // ID снапшота, который уже есть у клиента
int64 offset = 3; // Смещение в байтах для докачки
}
message DeleteSnapshotRequest {
// Запрос на получение информации о дифе
message GetDiffInfoRequest {
string snapshot_id = 1;
string local_parent_id = 2;
}
message DeleteSnapshotResponse {
bool success = 1;
// Информация о дифе
message DiffInfo {
string sha256_hash = 1;
int64 size_bytes = 2;
}
*/

View File

@@ -1,7 +1,7 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.5.1
// - protoc v4.25.3
// - protoc v6.32.0
// source: snapshot.proto
package grpc
@@ -22,6 +22,8 @@ const (
SnapshotService_ListSnapshots_FullMethodName = "/agate.grpc.SnapshotService/ListSnapshots"
SnapshotService_GetSnapshotDetails_FullMethodName = "/agate.grpc.SnapshotService/GetSnapshotDetails"
SnapshotService_DownloadFile_FullMethodName = "/agate.grpc.SnapshotService/DownloadFile"
SnapshotService_DownloadSnapshotDiff_FullMethodName = "/agate.grpc.SnapshotService/DownloadSnapshotDiff"
SnapshotService_GetDiffInfo_FullMethodName = "/agate.grpc.SnapshotService/GetDiffInfo"
)
// SnapshotServiceClient is the client API for SnapshotService service.
@@ -36,6 +38,10 @@ type SnapshotServiceClient interface {
GetSnapshotDetails(ctx context.Context, in *GetSnapshotDetailsRequest, opts ...grpc.CallOption) (*SnapshotDetails, error)
// Скачать конкретный файл из снапшота (потоковая передача)
DownloadFile(ctx context.Context, in *DownloadFileRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[DownloadFileResponse], error)
// Скачать архив, содержащий только разницу между двумя снапшотами
DownloadSnapshotDiff(ctx context.Context, in *DownloadSnapshotDiffRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[DownloadFileResponse], error)
// Получить информацию о дифе (хеш и размер)
GetDiffInfo(ctx context.Context, in *GetDiffInfoRequest, opts ...grpc.CallOption) (*DiffInfo, error)
}
type snapshotServiceClient struct {
@@ -85,6 +91,35 @@ func (c *snapshotServiceClient) DownloadFile(ctx context.Context, in *DownloadFi
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type SnapshotService_DownloadFileClient = grpc.ServerStreamingClient[DownloadFileResponse]
func (c *snapshotServiceClient) DownloadSnapshotDiff(ctx context.Context, in *DownloadSnapshotDiffRequest, opts ...grpc.CallOption) (grpc.ServerStreamingClient[DownloadFileResponse], error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
stream, err := c.cc.NewStream(ctx, &SnapshotService_ServiceDesc.Streams[1], SnapshotService_DownloadSnapshotDiff_FullMethodName, cOpts...)
if err != nil {
return nil, err
}
x := &grpc.GenericClientStream[DownloadSnapshotDiffRequest, DownloadFileResponse]{ClientStream: stream}
if err := x.ClientStream.SendMsg(in); err != nil {
return nil, err
}
if err := x.ClientStream.CloseSend(); err != nil {
return nil, err
}
return x, nil
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type SnapshotService_DownloadSnapshotDiffClient = grpc.ServerStreamingClient[DownloadFileResponse]
func (c *snapshotServiceClient) GetDiffInfo(ctx context.Context, in *GetDiffInfoRequest, opts ...grpc.CallOption) (*DiffInfo, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(DiffInfo)
err := c.cc.Invoke(ctx, SnapshotService_GetDiffInfo_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// SnapshotServiceServer is the server API for SnapshotService service.
// All implementations must embed UnimplementedSnapshotServiceServer
// for forward compatibility.
@@ -97,6 +132,10 @@ type SnapshotServiceServer interface {
GetSnapshotDetails(context.Context, *GetSnapshotDetailsRequest) (*SnapshotDetails, error)
// Скачать конкретный файл из снапшота (потоковая передача)
DownloadFile(*DownloadFileRequest, grpc.ServerStreamingServer[DownloadFileResponse]) error
// Скачать архив, содержащий только разницу между двумя снапшотами
DownloadSnapshotDiff(*DownloadSnapshotDiffRequest, grpc.ServerStreamingServer[DownloadFileResponse]) error
// Получить информацию о дифе (хеш и размер)
GetDiffInfo(context.Context, *GetDiffInfoRequest) (*DiffInfo, error)
mustEmbedUnimplementedSnapshotServiceServer()
}
@@ -116,6 +155,12 @@ func (UnimplementedSnapshotServiceServer) GetSnapshotDetails(context.Context, *G
func (UnimplementedSnapshotServiceServer) DownloadFile(*DownloadFileRequest, grpc.ServerStreamingServer[DownloadFileResponse]) error {
return status.Errorf(codes.Unimplemented, "method DownloadFile not implemented")
}
func (UnimplementedSnapshotServiceServer) DownloadSnapshotDiff(*DownloadSnapshotDiffRequest, grpc.ServerStreamingServer[DownloadFileResponse]) error {
return status.Errorf(codes.Unimplemented, "method DownloadSnapshotDiff not implemented")
}
func (UnimplementedSnapshotServiceServer) GetDiffInfo(context.Context, *GetDiffInfoRequest) (*DiffInfo, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetDiffInfo not implemented")
}
func (UnimplementedSnapshotServiceServer) mustEmbedUnimplementedSnapshotServiceServer() {}
func (UnimplementedSnapshotServiceServer) testEmbeddedByValue() {}
@@ -184,6 +229,35 @@ func _SnapshotService_DownloadFile_Handler(srv interface{}, stream grpc.ServerSt
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type SnapshotService_DownloadFileServer = grpc.ServerStreamingServer[DownloadFileResponse]
func _SnapshotService_DownloadSnapshotDiff_Handler(srv interface{}, stream grpc.ServerStream) error {
m := new(DownloadSnapshotDiffRequest)
if err := stream.RecvMsg(m); err != nil {
return err
}
return srv.(SnapshotServiceServer).DownloadSnapshotDiff(m, &grpc.GenericServerStream[DownloadSnapshotDiffRequest, DownloadFileResponse]{ServerStream: stream})
}
// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name.
type SnapshotService_DownloadSnapshotDiffServer = grpc.ServerStreamingServer[DownloadFileResponse]
func _SnapshotService_GetDiffInfo_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(GetDiffInfoRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(SnapshotServiceServer).GetDiffInfo(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: SnapshotService_GetDiffInfo_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(SnapshotServiceServer).GetDiffInfo(ctx, req.(*GetDiffInfoRequest))
}
return interceptor(ctx, in, info, handler)
}
// SnapshotService_ServiceDesc is the grpc.ServiceDesc for SnapshotService service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
@@ -199,6 +273,10 @@ var SnapshotService_ServiceDesc = grpc.ServiceDesc{
MethodName: "GetSnapshotDetails",
Handler: _SnapshotService_GetSnapshotDetails_Handler,
},
{
MethodName: "GetDiffInfo",
Handler: _SnapshotService_GetDiffInfo_Handler,
},
},
Streams: []grpc.StreamDesc{
{
@@ -206,6 +284,11 @@ var SnapshotService_ServiceDesc = grpc.ServiceDesc{
Handler: _SnapshotService_DownloadFile_Handler,
ServerStreams: true,
},
{
StreamName: "DownloadSnapshotDiff",
Handler: _SnapshotService_DownloadSnapshotDiff_Handler,
ServerStreams: true,
},
},
Metadata: "snapshot.proto",
}

View File

@@ -1,389 +1,142 @@
package agate
import (
"bytes"
"context"
"log"
"os"
"path/filepath"
"strings"
"testing"
"time"
"gitea.unprism.ru/KRBL/Agate/remote"
"gitea.unprism.ru/KRBL/Agate/store"
)
// TestGRPCServerClient tests the interaction between a gRPC server and client.
// It creates multiple snapshots with different content on the server,
// connects a client to the server, downloads the latest snapshot,
// and verifies the contents of the files.
func TestGRPCServerClient(t *testing.T) {
// Skip this test in short mode
// TestFullUpdateCycle tests a complete workflow: full download, then incremental update.
func TestFullUpdateCycle(t *testing.T) {
if testing.Short() {
t.Skip("Skipping gRPC server-client test in short mode")
t.Skip("Skipping full gRPC update cycle test in short mode")
}
// Create a temporary directory for the server
// --- 1. Setup Server ---
serverDir, err := os.MkdirTemp("", "agate-server-*")
if err != nil {
t.Fatalf("Failed to create server temp directory: %v", err)
}
defer os.RemoveAll(serverDir)
// Create a temporary directory for the client
clientDir, err := os.MkdirTemp("", "agate-client-*")
if err != nil {
t.Fatalf("Failed to create client temp directory: %v", err)
}
defer os.RemoveAll(clientDir)
// Create Agate options for the server
serverOptions := AgateOptions{
WorkDir: serverDir,
}
// Create Agate instance for the server
serverAgate, err := New(serverOptions)
serverAgate, err := New(AgateOptions{WorkDir: serverDir})
if err != nil {
t.Fatalf("Failed to create server Agate instance: %v", err)
}
defer serverAgate.Close()
// Create a data directory
dataDir := serverAgate.options.BlobStore.GetActiveDir()
if err := os.MkdirAll(dataDir, 0755); err != nil {
t.Fatalf("Failed to create data directory: %v", err)
// --- 2. Create Initial Snapshot (A) ---
dataDir := serverAgate.GetActiveDir()
if err := os.WriteFile(filepath.Join(dataDir, "file1.txt"), []byte("Version 1"), 0644); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(dataDir, "file2.txt"), []byte("Original Content"), 0644); err != nil {
t.Fatal(err)
}
// Create initial test files for the first snapshot
initialFiles := map[string]string{
filepath.Join(dataDir, "file1.txt"): "Initial content of file 1",
filepath.Join(dataDir, "file2.txt"): "Initial content of file 2",
filepath.Join(dataDir, "subdir", "file3.txt"): "Initial content of file 3",
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Create subdirectory
if err := os.MkdirAll(filepath.Join(dataDir, "subdir"), 0755); err != nil {
t.Fatalf("Failed to create subdirectory: %v", err)
}
// Create the files
for path, content := range initialFiles {
if err := os.WriteFile(path, []byte(content), 0644); err != nil {
t.Fatalf("Failed to create test file %s: %v", path, err)
}
}
// Create the first snapshot
ctx := context.Background()
snapshot1ID, err := serverAgate.SaveSnapshot(ctx, "Snapshot 1", "")
snapshotAID, err := serverAgate.SaveSnapshot(ctx, "Snapshot A", "")
if err != nil {
t.Fatalf("Failed to create first snapshot: %v", err)
t.Fatalf("Failed to create Snapshot A: %v", err)
}
t.Logf("Created first snapshot with ID: %s", snapshot1ID)
t.Logf("Created Snapshot A with ID: %s", snapshotAID)
// Modify some files and add a new file for the second snapshot
modifiedFiles := map[string]string{
filepath.Join(dataDir, "file1.txt"): "Modified content of file 1",
filepath.Join(dataDir, "file4.txt"): "Content of new file 4",
}
for path, content := range modifiedFiles {
if err := os.WriteFile(path, []byte(content), 0644); err != nil {
t.Fatalf("Failed to modify/create test file %s: %v", path, err)
}
}
// Create the second snapshot
snapshot2ID, err := serverAgate.SaveSnapshot(ctx, "Snapshot 2", snapshot1ID)
if err != nil {
t.Fatalf("Failed to create second snapshot: %v", err)
}
t.Logf("Created second snapshot with ID: %s", snapshot2ID)
// Delete a file and modify another for the third snapshot
if err := os.Remove(filepath.Join(dataDir, "file2.txt")); err != nil {
t.Fatalf("Failed to delete test file: %v", err)
}
if err := os.WriteFile(filepath.Join(dataDir, "subdir/file3.txt"), []byte("Modified content of file 3"), 0644); err != nil {
t.Fatalf("Failed to modify test file: %v", err)
}
// Create the third snapshot
snapshot3ID, err := serverAgate.SaveSnapshot(ctx, "Snapshot 3", snapshot2ID)
if err != nil {
t.Fatalf("Failed to create third snapshot: %v", err)
}
t.Logf("Created third snapshot with ID: %s", snapshot3ID)
// Start the gRPC server
// --- 3. Start Server ---
serverAddress := "localhost:50051"
server, err := remote.RunServer(ctx, serverAgate.manager, serverAddress)
if err != nil {
t.Fatalf("Failed to start gRPC server: %v", err)
server := remote.NewServer(serverAgate.manager)
go func() {
if err := server.Start(ctx, serverAddress); err != nil && err != context.Canceled {
log.Printf("Server start error: %v", err)
}
defer server.Stop(ctx)
// Give the server a moment to start
}()
defer server.Stop()
time.Sleep(100 * time.Millisecond)
// Connect a client to the server
client, err := remote.NewClient(serverAddress)
if err != nil {
t.Fatalf("Failed to connect client to server: %v", err)
}
defer client.Close()
// List snapshots from the client
snapshots, err := client.ListSnapshots(ctx)
if err != nil {
t.Fatalf("Failed to list snapshots from client: %v", err)
}
// Verify we have 3 snapshots
if len(snapshots) != 3 {
t.Errorf("Expected 3 snapshots, got %d", len(snapshots))
}
// Find the latest snapshot (should be snapshot3)
var latestSnapshot store.SnapshotInfo
for _, snapshot := range snapshots {
if latestSnapshot.CreationTime.Before(snapshot.CreationTime) {
latestSnapshot = snapshot
}
}
// Verify the latest snapshot is snapshot3
if latestSnapshot.ID != snapshot3ID {
t.Errorf("Latest snapshot ID is %s, expected %s", latestSnapshot.ID, snapshot3ID)
}
// Get detailed information about the latest snapshot
snapshotDetails, err := client.FetchSnapshotDetails(ctx, latestSnapshot.ID)
if err != nil {
t.Fatalf("Failed to fetch snapshot details: %v", err)
}
// Verify the snapshot details
if snapshotDetails.ID != snapshot3ID {
t.Errorf("Snapshot details ID is %s, expected %s", snapshotDetails.ID, snapshot3ID)
}
// Create a directory to download the snapshot to
downloadDir := filepath.Join(clientDir, "download")
if err := os.MkdirAll(downloadDir, 0755); err != nil {
t.Fatalf("Failed to create download directory: %v", err)
}
// Download the snapshot
err = client.DownloadSnapshot(ctx, latestSnapshot.ID, downloadDir, "")
if err != nil {
t.Fatalf("Failed to download snapshot: %v", err)
}
// Verify the downloaded files match the expected content
expectedFiles := map[string]string{
filepath.Join(downloadDir, "file1.txt"): "Modified content of file 1",
filepath.Join(downloadDir, "file4.txt"): "Content of new file 4",
filepath.Join(downloadDir, "subdir/file3.txt"): "Modified content of file 3",
}
for path, expectedContent := range expectedFiles {
content, err := os.ReadFile(path)
if err != nil {
t.Fatalf("Failed to read downloaded file %s: %v", path, err)
}
if string(content) != expectedContent {
t.Errorf("Downloaded file %s has wrong content: got %s, want %s", path, string(content), expectedContent)
}
}
// Verify that file2.txt doesn't exist in the downloaded snapshot
if _, err := os.Stat(filepath.Join(downloadDir, "file2.txt")); !os.IsNotExist(err) {
t.Errorf("file2.txt should not exist in the downloaded snapshot")
}
}
// TestGRPC_GetRemoteSnapshot_Incremental tests the incremental download functionality
// of GetRemoteSnapshot, verifying that it reuses files from a local parent snapshot
// instead of downloading them again.
func TestGRPC_GetRemoteSnapshot_Incremental(t *testing.T) {
// Skip this test in short mode
if testing.Short() {
t.Skip("Skipping incremental GetRemoteSnapshot test in short mode")
}
// Create a temporary directory for the server
serverDir, err := os.MkdirTemp("", "agate-server-*")
if err != nil {
t.Fatalf("Failed to create server temp directory: %v", err)
}
defer os.RemoveAll(serverDir)
// Create a temporary directory for the client
// --- 4. Setup Client ---
clientDir, err := os.MkdirTemp("", "agate-client-*")
if err != nil {
t.Fatalf("Failed to create client temp directory: %v", err)
}
defer os.RemoveAll(clientDir)
// Create a buffer to capture client logs
var clientLogBuffer bytes.Buffer
clientLogger := log.New(&clientLogBuffer, "", 0)
// Create Agate options for the server
serverOptions := AgateOptions{
WorkDir: serverDir,
}
// Create Agate options for the client with logger
clientOptions := AgateOptions{
WorkDir: clientDir,
Logger: clientLogger,
}
// Create Agate instances for server and client
serverAgate, err := New(serverOptions)
if err != nil {
t.Fatalf("Failed to create server Agate instance: %v", err)
}
defer serverAgate.Close()
clientAgate, err := New(clientOptions)
clientAgate, err := New(AgateOptions{WorkDir: clientDir, CleanOnRestore: true})
if err != nil {
t.Fatalf("Failed to create client Agate instance: %v", err)
}
defer clientAgate.Close()
// Create a data directory on the server
serverDataDir := serverAgate.options.BlobStore.GetActiveDir()
if err := os.MkdirAll(serverDataDir, 0755); err != nil {
t.Fatalf("Failed to create server data directory: %v", err)
// --- 5. Client Performs Full Download of Snapshot A ---
t.Log("Client performing full download of Snapshot A...")
if err := clientAgate.GetRemoteSnapshot(ctx, serverAddress, snapshotAID, ""); err != nil {
t.Fatalf("Client failed to get Snapshot A: %v", err)
}
// Create test files for snapshot A on the server
if err := os.MkdirAll(filepath.Join(serverDataDir, "subdir"), 0755); err != nil {
t.Fatalf("Failed to create subdirectory: %v", err)
// Verify content of Snapshot A on client
if err := clientAgate.RestoreSnapshot(ctx, snapshotAID); err != nil {
t.Fatalf("Failed to restore Snapshot A: %v", err)
}
verifyFileContent(t, clientAgate.GetActiveDir(), "file1.txt", "Version 1")
verifyFileContent(t, clientAgate.GetActiveDir(), "file2.txt", "Original Content")
t.Log("Snapshot A verified on client.")
// --- 6. Server Creates Incremental Snapshot (B) ---
if err := os.WriteFile(filepath.Join(dataDir, "file1.txt"), []byte("Version 2"), 0644); err != nil {
t.Fatal(err)
}
if err := os.WriteFile(filepath.Join(dataDir, "file3.txt"), []byte("New File"), 0644); err != nil {
t.Fatal(err)
}
if err := os.Remove(filepath.Join(dataDir, "file2.txt")); err != nil {
t.Fatal(err)
}
snapshotAFiles := map[string]string{
filepath.Join(serverDataDir, "file1.txt"): "Content of file 1",
filepath.Join(serverDataDir, "file2.txt"): "Content of file 2",
filepath.Join(serverDataDir, "subdir/file3.txt"): "Content of file 3",
}
for path, content := range snapshotAFiles {
if err := os.WriteFile(path, []byte(content), 0644); err != nil {
t.Fatalf("Failed to create test file %s: %v", path, err)
}
}
// Create snapshot A on the server
ctx := context.Background()
snapshotAID, err := serverAgate.SaveSnapshot(ctx, "Snapshot A", "")
if err != nil {
t.Fatalf("Failed to create snapshot A: %v", err)
}
t.Logf("Created snapshot A with ID: %s", snapshotAID)
// Modify some files and add a new file for snapshot B
snapshotBChanges := map[string]string{
filepath.Join(serverDataDir, "file1.txt"): "Modified content of file 1", // Modified file
filepath.Join(serverDataDir, "file4.txt"): "Content of new file 4", // New file
filepath.Join(serverDataDir, "subdir/file5.txt"): "Content of new file 5", // New file in subdir
}
for path, content := range snapshotBChanges {
// Ensure parent directory exists
if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil {
t.Fatalf("Failed to create directory for %s: %v", path, err)
}
if err := os.WriteFile(path, []byte(content), 0644); err != nil {
t.Fatalf("Failed to create/modify test file %s: %v", path, err)
}
}
// Create snapshot B on the server (with A as parent)
snapshotBID, err := serverAgate.SaveSnapshot(ctx, "Snapshot B", snapshotAID)
if err != nil {
t.Fatalf("Failed to create snapshot B: %v", err)
t.Fatalf("Failed to create Snapshot B: %v", err)
}
t.Logf("Created snapshot B with ID: %s", snapshotBID)
t.Logf("Created Snapshot B with ID: %s", snapshotBID)
// Start the gRPC server
serverAddress := "localhost:50052" // Use a different port than the other test
server, err := remote.RunServer(ctx, serverAgate.manager, serverAddress)
// --- 7. Client Performs Incremental Download of Snapshot B ---
t.Log("Client performing incremental download of Snapshot B...")
parentIDOnClient := clientAgate.GetCurrentSnapshotID()
if parentIDOnClient != snapshotAID {
t.Fatalf("Client has incorrect current snapshot ID. Got %s, want %s", parentIDOnClient, snapshotAID)
}
if err := clientAgate.GetRemoteSnapshot(ctx, serverAddress, snapshotBID, parentIDOnClient); err != nil {
t.Fatalf("Client failed to get Snapshot B: %v", err)
}
// --- 8. Verify Final State on Client ---
if err := clientAgate.RestoreSnapshot(ctx, snapshotBID); err != nil {
t.Fatalf("Failed to restore Snapshot B: %v", err)
}
t.Log("Snapshot B restored on client. Verifying content...")
verifyFileContent(t, clientAgate.GetActiveDir(), "file1.txt", "Version 2")
verifyFileContent(t, clientAgate.GetActiveDir(), "file3.txt", "New File")
// Verify that file2.txt was removed because CleanOnRestore is true
if _, err := os.Stat(filepath.Join(clientAgate.GetActiveDir(), "file2.txt")); !os.IsNotExist(err) {
t.Errorf("file2.txt should have been removed after restoring Snapshot B, but it still exists.")
}
t.Log("Final state verified successfully!")
}
func verifyFileContent(t *testing.T, dir, filename, expectedContent string) {
t.Helper()
content, err := os.ReadFile(filepath.Join(dir, filename))
if err != nil {
t.Fatalf("Failed to start gRPC server: %v", err)
}
defer server.Stop(ctx)
// Give the server a moment to start
time.Sleep(100 * time.Millisecond)
// Step 1: Client downloads snapshot A
err = clientAgate.GetRemoteSnapshot(ctx, serverAddress, snapshotAID, "")
if err != nil {
t.Fatalf("Failed to download snapshot A: %v", err)
}
t.Log("Client successfully downloaded snapshot A")
// Clear the log buffer to capture only logs from the incremental download
clientLogBuffer.Reset()
// Step 2: Client downloads snapshot B, specifying A as the local parent
err = clientAgate.GetRemoteSnapshot(ctx, serverAddress, snapshotBID, snapshotAID)
if err != nil {
t.Fatalf("Failed to download snapshot B: %v", err)
}
t.Log("Client successfully downloaded snapshot B")
// Step 3: Verify that snapshot B was correctly imported
// Restore snapshot B to a directory
restoreDir := filepath.Join(clientDir, "restore")
if err := os.MkdirAll(restoreDir, 0755); err != nil {
t.Fatalf("Failed to create restore directory: %v", err)
}
err = clientAgate.RestoreSnapshotToDir(ctx, snapshotBID, restoreDir)
if err != nil {
t.Fatalf("Failed to restore snapshot B: %v", err)
}
// Verify the restored files match the expected content
expectedFiles := map[string]string{
filepath.Join(restoreDir, "file1.txt"): "Modified content of file 1", // Modified file
filepath.Join(restoreDir, "file2.txt"): "Content of file 2", // Unchanged file
filepath.Join(restoreDir, "file4.txt"): "Content of new file 4", // New file
filepath.Join(restoreDir, "subdir/file3.txt"): "Content of file 3", // Unchanged file
filepath.Join(restoreDir, "subdir/file5.txt"): "Content of new file 5", // New file
}
for path, expectedContent := range expectedFiles {
content, err := os.ReadFile(path)
if err != nil {
t.Fatalf("Failed to read restored file %s: %v", path, err)
t.Fatalf("Failed to read file %s: %v", filename, err)
}
if string(content) != expectedContent {
t.Errorf("Restored file %s has wrong content: got %s, want %s", path, string(content), expectedContent)
t.Errorf("File %s has wrong content: got '%s', want '%s'", filename, string(content), expectedContent)
}
}
// Step 4: Analyze logs to verify incremental download behavior
logs := clientLogBuffer.String()
// Check for evidence of file reuse
if !strings.Contains(logs, "Reusing file") {
t.Errorf("No evidence of file reuse in logs")
}
// Check for evidence of downloading only new/changed files
if !strings.Contains(logs, "Downloading file") {
t.Errorf("No evidence of downloading new files in logs")
}
// Log the relevant parts for debugging
t.Logf("Log evidence of incremental download:\n%s", logs)
}

View File

@@ -13,6 +13,15 @@ type SnapshotManager interface {
// Returns the created Snapshot with its metadata or an error if the process fails.
CreateSnapshot(ctx context.Context, sourceDir string, name string, parentID string) (*store.Snapshot, error)
// CreateSnapshotAsync initiates a background process to create a snapshot.
// Returns the job ID (which is also the snapshot ID) or an error if the process couldn't start.
// onStart is called in the background goroutine before the snapshot creation starts.
// onFinish is called in the background goroutine after the snapshot creation finishes (successfully or with error).
CreateSnapshotAsync(ctx context.Context, sourceDir string, name string, parentID string, onStart func(), onFinish func(string, error)) (string, error)
// GetSnapshotStatus retrieves the status of an asynchronous snapshot creation job.
GetSnapshotStatus(ctx context.Context, jobID string) (*store.SnapshotStatus, error)
// GetSnapshotDetails retrieves detailed metadata for a specific snapshot identified by its unique snapshotID.
// Returns a Snapshot object containing metadata
GetSnapshotDetails(ctx context.Context, snapshotID string) (*store.Snapshot, error)
@@ -33,6 +42,14 @@ type SnapshotManager interface {
// UpdateSnapshotMetadata updates the metadata of an existing snapshot, allowing changes to its name.
UpdateSnapshotMetadata(ctx context.Context, snapshotID string, newName string) error
// StreamSnapshotDiff creates and streams a differential archive between two snapshots.
// It returns an io.ReadCloser for the archive stream and an error.
// The caller is responsible for closing the reader, which will also handle cleanup of temporary resources.
StreamSnapshotDiff(ctx context.Context, snapshotID, parentID string, offset int64) (io.ReadCloser, error)
// GetSnapshotDiffInfo calculates the hash and size of a differential archive between two snapshots.
GetSnapshotDiffInfo(ctx context.Context, snapshotID, parentID string) (*store.DiffInfo, error)
}
// SnapshotServer defines the interface for a server that can share snapshots
@@ -52,8 +69,8 @@ type SnapshotClient interface {
// FetchSnapshotDetails retrieves detailed information about a specific snapshot
FetchSnapshotDetails(ctx context.Context, snapshotID string) (*store.Snapshot, error)
// DownloadSnapshot downloads a snapshot from the server
DownloadSnapshot(ctx context.Context, snapshotID string, targetDir string, localParentID string) error
// DownloadSnapshotDiff downloads a differential archive between two snapshots to a target directory
DownloadSnapshotDiff(ctx context.Context, snapshotID, localParentID, targetPath string) error
// Close closes the connection to the server
Close() error

View File

@@ -2,6 +2,7 @@ package agate
import (
"archive/zip"
"bytes"
"context"
"errors"
"fmt"
@@ -10,8 +11,10 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"time"
"gitea.unprism.ru/KRBL/Agate/models"
"github.com/google/uuid"
"gitea.unprism.ru/KRBL/Agate/archive"
@@ -24,6 +27,8 @@ type SnapshotManagerData struct {
metadataStore store.MetadataStore
blobStore store.BlobStore
logger *log.Logger
jobs map[string]*store.SnapshotStatus
jobsMutex sync.RWMutex
}
func CreateSnapshotManager(metadataStore store.MetadataStore, blobStore store.BlobStore, logger *log.Logger) (interfaces.SnapshotManager, error) {
@@ -31,10 +36,16 @@ func CreateSnapshotManager(metadataStore store.MetadataStore, blobStore store.Bl
return nil, errors.New("parameters can't be nil")
}
// Ensure logger is never nil.
if logger == nil {
logger = log.New(io.Discard, "", 0)
}
return &SnapshotManagerData{
metadataStore: metadataStore,
blobStore: blobStore,
logger: logger,
jobs: make(map[string]*store.SnapshotStatus),
}, nil
}
@@ -43,13 +54,13 @@ func (data *SnapshotManagerData) CreateSnapshot(ctx context.Context, sourceDir s
info, err := os.Stat(sourceDir)
if err != nil {
if os.IsNotExist(err) {
return nil, ErrSourceNotFound
return nil, models.ErrSourceNotFound
}
return nil, fmt.Errorf("failed to access source directory: %w", err)
}
if !info.IsDir() {
return nil, ErrSourceNotDirectory
return nil, models.ErrSourceNotDirectory
}
// Check if parent exists if specified
@@ -64,6 +75,98 @@ func (data *SnapshotManagerData) CreateSnapshot(ctx context.Context, sourceDir s
// Generate a unique ID for the snapshot
snapshotID := uuid.New().String()
return data.createSnapshotInternal(ctx, sourceDir, name, parentID, snapshotID, nil)
}
func (data *SnapshotManagerData) CreateSnapshotAsync(ctx context.Context, sourceDir string, name string, parentID string, onStart func(), onFinish func(string, error)) (string, error) {
// Validate source directory
info, err := os.Stat(sourceDir)
if err != nil {
if os.IsNotExist(err) {
return "", models.ErrSourceNotFound
}
return "", fmt.Errorf("failed to access source directory: %w", err)
}
if !info.IsDir() {
return "", models.ErrSourceNotDirectory
}
// Check if parent exists if specified
if parentID != "" {
_, err := data.metadataStore.GetSnapshotMetadata(ctx, parentID)
if err != nil {
fmt.Println("failed to check parent snapshot: %w", err)
parentID = ""
}
}
snapshotID := uuid.New().String()
data.jobsMutex.Lock()
data.jobs[snapshotID] = &store.SnapshotStatus{
ID: snapshotID,
Status: "pending",
Progress: 0,
}
data.jobsMutex.Unlock()
go func() {
if onStart != nil {
onStart()
}
data.jobsMutex.Lock()
if job, ok := data.jobs[snapshotID]; ok {
job.Status = "running"
}
data.jobsMutex.Unlock()
_, err := data.createSnapshotInternal(context.Background(), sourceDir, name, parentID, snapshotID, func(current, total int64) {
data.jobsMutex.Lock()
defer data.jobsMutex.Unlock()
if job, ok := data.jobs[snapshotID]; ok {
if total > 0 {
job.Progress = float64(current) / float64(total)
}
}
})
data.jobsMutex.Lock()
if job, ok := data.jobs[snapshotID]; ok {
if err != nil {
job.Status = "failed"
job.Error = err.Error()
} else {
job.Status = "done"
job.Progress = 1.0
}
}
data.jobsMutex.Unlock()
if onFinish != nil {
onFinish(snapshotID, err)
}
}()
return snapshotID, nil
}
func (data *SnapshotManagerData) GetSnapshotStatus(ctx context.Context, jobID string) (*store.SnapshotStatus, error) {
data.jobsMutex.RLock()
defer data.jobsMutex.RUnlock()
job, ok := data.jobs[jobID]
if !ok {
return nil, models.ErrNotFound
}
// Return a copy to avoid race conditions
statusCopy := *job
return &statusCopy, nil
}
func (data *SnapshotManagerData) createSnapshotInternal(ctx context.Context, sourceDir, name, parentID, snapshotID string, onProgress func(current, total int64)) (*store.Snapshot, error) {
// Create a temporary file for the archive in the working directory
tempFilePath := filepath.Join(data.blobStore.GetBaseDir(), "temp-"+snapshotID+".zip")
tempFile, err := os.Create(tempFilePath)
@@ -74,7 +177,7 @@ func (data *SnapshotManagerData) CreateSnapshot(ctx context.Context, sourceDir s
defer os.Remove(tempFilePath) // Clean up temp file after we're done
// Create archive of the source directory
if err := archive.CreateArchive(sourceDir, tempFilePath); err != nil {
if err := archive.CreateArchiveWithProgress(sourceDir, tempFilePath, onProgress); err != nil {
return nil, fmt.Errorf("failed to create archive: %w", err)
}
@@ -159,8 +262,8 @@ func (data *SnapshotManagerData) GetSnapshotDetails(ctx context.Context, snapsho
// Retrieve snapshot metadata from the store
snapshot, err := data.metadataStore.GetSnapshotMetadata(ctx, snapshotID)
if err != nil {
if errors.Is(err, ErrNotFound) {
return nil, ErrNotFound
if errors.Is(err, models.ErrNotFound) {
return nil, models.ErrNotFound
}
return nil, fmt.Errorf("failed to retrieve snapshot details: %w", err)
}
@@ -183,31 +286,24 @@ func (data *SnapshotManagerData) DeleteSnapshot(ctx context.Context, snapshotID
return errors.New("snapshot ID cannot be empty")
}
// First check if the snapshot exists and get its details
snapshot, err := data.metadataStore.GetSnapshotMetadata(ctx, snapshotID)
if err != nil {
if errors.Is(err, ErrNotFound) {
// If snapshot doesn't exist, return success (idempotent operation)
if errors.Is(err, models.ErrNotFound) {
return nil
}
return fmt.Errorf("failed to check if snapshot exists: %w", err)
}
// Get the parent ID of the snapshot being deleted
parentID := snapshot.ParentID
// Find all snapshots that have the deleted snapshot as their parent
// We need to update their parent ID to maintain the chain
opts := store.ListOptions{}
allSnapshots, err := data.metadataStore.ListSnapshotsMetadata(ctx, opts)
if err != nil {
return fmt.Errorf("failed to list snapshots: %w", err)
}
// Update parent references for any snapshots that have this one as a parent
for _, info := range allSnapshots {
if info.ParentID == snapshotID {
// Используем новый, более надежный метод для обновления только parent_id
if err := data.metadataStore.UpdateSnapshotParentID(ctx, info.ID, parentID); err != nil {
data.logger.Printf("WARNING: failed to update parent reference for snapshot %s: %v", info.ID, err)
} else {
@@ -216,16 +312,11 @@ func (data *SnapshotManagerData) DeleteSnapshot(ctx context.Context, snapshotID
}
}
// Delete the metadata first
if err := data.metadataStore.DeleteSnapshotMetadata(ctx, snapshotID); err != nil {
return fmt.Errorf("failed to delete snapshot metadata: %w", err)
}
// Then delete the blob
if err := data.blobStore.DeleteBlob(ctx, snapshotID); err != nil {
// Note: We don't return here because we've already deleted the metadata
// and the blob store should handle the case where the blob doesn't exist
// Log the error instead
data.logger.Printf("WARNING: failed to delete snapshot blob: %v", err)
}
@@ -243,8 +334,8 @@ func (data *SnapshotManagerData) OpenFile(ctx context.Context, snapshotID string
// First check if the snapshot exists
_, err := data.metadataStore.GetSnapshotMetadata(ctx, snapshotID)
if err != nil {
if errors.Is(err, ErrNotFound) {
return nil, ErrNotFound
if errors.Is(err, models.ErrNotFound) {
return nil, models.ErrNotFound
}
return nil, fmt.Errorf("failed to check if snapshot exists: %w", err)
}
@@ -267,7 +358,7 @@ func (data *SnapshotManagerData) OpenFile(ctx context.Context, snapshotID string
err := archive.ExtractFileFromArchive(blobPath, filePath, pw)
if err != nil {
if errors.Is(err, archive.ErrFileNotFoundInArchive) {
pw.CloseWithError(ErrFileNotFound)
pw.CloseWithError(models.ErrFileNotFound)
return
}
pw.CloseWithError(fmt.Errorf("failed to extract file from archive: %w", err))
@@ -290,8 +381,8 @@ func (data *SnapshotManagerData) ExtractSnapshot(ctx context.Context, snapshotID
// First check if the snapshot exists
_, err := data.metadataStore.GetSnapshotMetadata(ctx, snapshotID)
if err != nil {
if errors.Is(err, ErrNotFound) {
return ErrNotFound
if errors.Is(err, models.ErrNotFound) {
return models.ErrNotFound
}
return fmt.Errorf("failed to check if snapshot exists: %w", err)
}
@@ -392,22 +483,178 @@ func (data *SnapshotManagerData) UpdateSnapshotMetadata(ctx context.Context, sna
return errors.New("new name cannot be empty")
}
// Get the current snapshot metadata
snapshot, err := data.metadataStore.GetSnapshotMetadata(ctx, snapshotID)
if err != nil {
if errors.Is(err, ErrNotFound) {
return ErrNotFound
if errors.Is(err, models.ErrNotFound) {
return models.ErrNotFound
}
return fmt.Errorf("failed to get snapshot metadata: %w", err)
}
// Update the name
snapshot.Name = newName
// Save the updated metadata
if err := data.metadataStore.SaveSnapshotMetadata(ctx, *snapshot); err != nil {
return fmt.Errorf("failed to update snapshot metadata: %w", err)
}
return nil
}
func (data *SnapshotManagerData) GetSnapshotDiffInfo(ctx context.Context, snapshotID, parentID string) (*store.DiffInfo, error) {
tempArchivePath, tempStagingDir, err := data.createDiffArchive(ctx, snapshotID, parentID)
if err != nil {
return nil, fmt.Errorf("failed to create diff archive for info: %w", err)
}
if tempArchivePath == "" {
return &store.DiffInfo{SHA256: "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", Size: 0}, nil // sha256 of empty string
}
defer os.Remove(tempArchivePath)
if tempStagingDir != "" {
defer os.RemoveAll(tempStagingDir)
}
hash, err := hash.CalculateFileHash(tempArchivePath)
if err != nil {
return nil, fmt.Errorf("failed to calculate hash for diff archive: %w", err)
}
stat, err := os.Stat(tempArchivePath)
if err != nil {
return nil, fmt.Errorf("failed to get size of diff archive: %w", err)
}
return &store.DiffInfo{
SHA256: hash,
Size: stat.Size(),
}, nil
}
// diffArchiveReader is a wrapper around an *os.File that handles cleanup of temporary files.
type diffArchiveReader struct {
*os.File
tempArchive string
tempStaging string
}
// Close closes the file and removes the temporary archive and staging directory.
func (r *diffArchiveReader) Close() error {
err := r.File.Close()
_ = os.Remove(r.tempArchive)
_ = os.RemoveAll(r.tempStaging)
return err
}
func (data *SnapshotManagerData) createDiffArchive(ctx context.Context, snapshotID, parentID string) (string, string, error) {
targetSnap, err := data.metadataStore.GetSnapshotMetadata(ctx, snapshotID)
if err != nil {
return "", "", fmt.Errorf("failed to get target snapshot metadata: %w", err)
}
parentFiles := make(map[string]string)
if parentID != "" {
parentSnap, err := data.metadataStore.GetSnapshotMetadata(ctx, parentID)
if err == nil {
for _, file := range parentSnap.Files {
if !file.IsDir {
parentFiles[file.Path] = file.SHA256
}
}
} else {
data.logger.Printf("Warning: failed to get parent snapshot %s, creating full diff: %v", parentID, err)
}
}
var filesToInclude []string
for _, file := range targetSnap.Files {
if file.IsDir {
continue
}
if parentHash, ok := parentFiles[file.Path]; !ok || parentHash != file.SHA256 {
filesToInclude = append(filesToInclude, file.Path)
}
}
if len(filesToInclude) == 0 {
return "", "", nil
}
tempStagingDir, err := os.MkdirTemp(data.blobStore.GetBaseDir(), "diff-staging-*")
if err != nil {
return "", "", fmt.Errorf("failed to create temp staging directory: %w", err)
}
targetBlobPath, err := data.blobStore.GetBlobPath(ctx, snapshotID)
if err != nil {
os.RemoveAll(tempStagingDir)
return "", "", err
}
for _, filePath := range filesToInclude {
destPath := filepath.Join(tempStagingDir, filePath)
if err := os.MkdirAll(filepath.Dir(destPath), 0755); err != nil {
os.RemoveAll(tempStagingDir)
return "", "", fmt.Errorf("failed to create dir for diff file: %w", err)
}
fileWriter, err := os.Create(destPath)
if err != nil {
os.RemoveAll(tempStagingDir)
return "", "", err
}
err = archive.ExtractFileFromArchive(targetBlobPath, filePath, fileWriter)
fileWriter.Close()
if err != nil {
os.RemoveAll(tempStagingDir)
return "", "", fmt.Errorf("failed to extract file %s for diff: %w", filePath, err)
}
}
tempArchivePath := filepath.Join(data.blobStore.GetBaseDir(), "diff-"+snapshotID+".zip")
if err := archive.CreateArchive(tempStagingDir, tempArchivePath); err != nil {
_ = os.RemoveAll(tempStagingDir)
_ = os.Remove(tempArchivePath)
return "", "", fmt.Errorf("failed to create diff archive: %w", err)
}
return tempArchivePath, tempStagingDir, nil
}
func (data *SnapshotManagerData) StreamSnapshotDiff(ctx context.Context, snapshotID, parentID string, offset int64) (io.ReadCloser, error) {
tempArchivePath, tempStagingDir, err := data.createDiffArchive(ctx, snapshotID, parentID)
if err != nil {
return nil, fmt.Errorf("failed to create diff archive for streaming: %w", err)
}
if tempArchivePath == "" {
return io.NopCloser(bytes.NewReader(nil)), nil
}
archiveFile, err := os.Open(tempArchivePath)
if err != nil {
if tempStagingDir != "" {
os.RemoveAll(tempStagingDir)
}
os.Remove(tempArchivePath)
return nil, err
}
if offset > 0 {
if _, err := archiveFile.Seek(offset, io.SeekStart); err != nil {
archiveFile.Close()
if tempStagingDir != "" {
os.RemoveAll(tempStagingDir)
}
os.Remove(tempArchivePath)
return nil, fmt.Errorf("failed to seek in diff archive: %w", err)
}
}
return &diffArchiveReader{
File: archiveFile,
tempArchive: tempArchivePath,
tempStaging: tempStagingDir,
}, nil
}

View File

@@ -554,3 +554,60 @@ func TestUpdateSnapshotMetadata(t *testing.T) {
t.Fatalf("Expected error when updating non-existent snapshot, got nil")
}
}
func TestStreamSnapshotDiff_EdgeCases(t *testing.T) {
tempDir, metadataStore, blobStore, cleanup := setupTestEnvironment(t)
defer cleanup()
sourceDir := filepath.Join(tempDir, "source")
os.MkdirAll(sourceDir, 0755)
createTestFiles(t, sourceDir)
manager, err := CreateSnapshotManager(metadataStore, blobStore, nil)
if err != nil {
t.Fatalf("Failed to create snapshot manager: %v", err)
}
ctx := context.Background()
// Create two identical snapshots
snap1, _ := manager.CreateSnapshot(ctx, sourceDir, "Snap1", "")
snap2, _ := manager.CreateSnapshot(ctx, sourceDir, "Snap2", snap1.ID)
// Test 1: Diff between identical snapshots should be empty
reader, err := manager.StreamSnapshotDiff(ctx, snap2.ID, snap1.ID, 0)
if err != nil {
t.Fatalf("Expected no error for identical snapshots, got %v", err)
}
defer reader.Close()
data, _ := io.ReadAll(reader)
if len(data) != 0 {
t.Errorf("Expected empty diff for identical snapshots, got %d bytes", len(data))
}
// Test 2: Diff with a non-existent parent should be a full archive
reader, err = manager.StreamSnapshotDiff(ctx, snap1.ID, "non-existent-parent", 0)
if err != nil {
t.Fatalf("Expected no error for non-existent parent, got %v", err)
}
defer reader.Close()
data, _ = io.ReadAll(reader)
if len(data) == 0 {
t.Error("Expected full archive for non-existent parent, got empty diff")
}
// Create an empty source dir
emptyDir := filepath.Join(tempDir, "empty_source")
os.MkdirAll(emptyDir, 0755)
emptySnap, _ := manager.CreateSnapshot(ctx, emptyDir, "EmptySnap", "")
// Test 3: Diff of an empty snapshot should be empty
reader, err = manager.StreamSnapshotDiff(ctx, emptySnap.ID, "", 0)
if err != nil {
t.Fatalf("Expected no error for empty snapshot, got %v", err)
}
defer reader.Close()
data, _ = io.ReadAll(reader)
if len(data) != 0 {
t.Errorf("Expected empty diff for empty snapshot, got %d bytes", len(data))
}
}

View File

@@ -1,4 +1,4 @@
package agate
package models
import "errors"

View File

@@ -11,38 +11,31 @@ import (
"google.golang.org/grpc/credentials/insecure"
agateGrpc "gitea.unprism.ru/KRBL/Agate/grpc"
"gitea.unprism.ru/KRBL/Agate/hash"
"gitea.unprism.ru/KRBL/Agate/interfaces"
"gitea.unprism.ru/KRBL/Agate/store"
)
// Client represents a client for connecting to a remote snapshot server
// It implements the interfaces.SnapshotClient interface
// Client представляет клиент для подключения к удаленному серверу снапшотов.
type Client struct {
conn *stdgrpc.ClientConn
client agateGrpc.SnapshotServiceClient
}
// Ensure Client implements interfaces.SnapshotClient
// Убедимся, что Client реализует интерфейс interfaces.SnapshotClient
var _ interfaces.SnapshotClient = (*Client)(nil)
// NewClient creates a new client connected to the specified address
// NewClient создает нового клиента, подключенного к указанному адресу.
func NewClient(address string) (*Client, error) {
// Connect to the server with insecure credentials (for simplicity)
conn, err := stdgrpc.Dial(address, stdgrpc.WithTransportCredentials(insecure.NewCredentials()))
if err != nil {
return nil, fmt.Errorf("failed to connect to server at %s: %w", address, err)
}
// Create the gRPC client
client := agateGrpc.NewSnapshotServiceClient(conn)
return &Client{
conn: conn,
client: client,
}, nil
return &Client{conn: conn, client: client}, nil
}
// Close closes the connection to the server
// Close закрывает соединение с сервером.
func (c *Client) Close() error {
if c.conn != nil {
return c.conn.Close()
@@ -50,14 +43,13 @@ func (c *Client) Close() error {
return nil
}
// ListSnapshots retrieves a list of snapshots from the remote server
// ListSnapshots получает список снапшотов с удаленного сервера.
func (c *Client) ListSnapshots(ctx context.Context) ([]store.SnapshotInfo, error) {
response, err := c.client.ListSnapshots(ctx, &agateGrpc.ListSnapshotsRequest{})
if err != nil {
return nil, fmt.Errorf("failed to list snapshots: %w", err)
}
// Convert gRPC snapshot info to store.SnapshotInfo
snapshots := make([]store.SnapshotInfo, 0, len(response.Snapshots))
for _, snapshot := range response.Snapshots {
snapshots = append(snapshots, store.SnapshotInfo{
@@ -67,11 +59,10 @@ func (c *Client) ListSnapshots(ctx context.Context) ([]store.SnapshotInfo, error
CreationTime: snapshot.CreationTime.AsTime(),
})
}
return snapshots, nil
}
// FetchSnapshotDetails retrieves detailed information about a specific snapshot
// FetchSnapshotDetails получает детальную информацию о конкретном снапшоте.
func (c *Client) FetchSnapshotDetails(ctx context.Context, snapshotID string) (*store.Snapshot, error) {
response, err := c.client.GetSnapshotDetails(ctx, &agateGrpc.GetSnapshotDetailsRequest{
SnapshotId: snapshotID,
@@ -80,7 +71,6 @@ func (c *Client) FetchSnapshotDetails(ctx context.Context, snapshotID string) (*
return nil, fmt.Errorf("failed to get snapshot details: %w", err)
}
// Convert gRPC snapshot details to store.Snapshot
snapshot := &store.Snapshot{
ID: response.Info.Id,
Name: response.Info.Name,
@@ -89,7 +79,6 @@ func (c *Client) FetchSnapshotDetails(ctx context.Context, snapshotID string) (*
Files: make([]store.FileInfo, 0, len(response.Files)),
}
// Convert file info
for _, file := range response.Files {
snapshot.Files = append(snapshot.Files, store.FileInfo{
Path: file.Path,
@@ -98,118 +87,87 @@ func (c *Client) FetchSnapshotDetails(ctx context.Context, snapshotID string) (*
SHA256: file.Sha256Hash,
})
}
return snapshot, nil
}
// DownloadSnapshot downloads a snapshot from the server
// This implementation downloads each file individually to optimize bandwidth usage
func (c *Client) DownloadSnapshot(ctx context.Context, snapshotID string, targetDir string, localParentID string) error {
// Get snapshot details
snapshot, err := c.FetchSnapshotDetails(ctx, snapshotID)
if err != nil {
return fmt.Errorf("failed to get snapshot details: %w", err)
}
// Create target directory if it doesn't exist
if err := os.MkdirAll(targetDir, 0755); err != nil {
return fmt.Errorf("failed to create target directory: %w", err)
}
// If a local parent is specified, get its details to compare files
var localParentFiles map[string]store.FileInfo
if localParentID != "" {
localParent, err := c.FetchSnapshotDetails(ctx, localParentID)
if err == nil {
// Create a map of file paths to file info for quick lookup
localParentFiles = make(map[string]store.FileInfo, len(localParent.Files))
for _, file := range localParent.Files {
localParentFiles[file.Path] = file
}
}
}
// Download each file
for _, file := range snapshot.Files {
// Skip directories, we'll create them when needed
if file.IsDir {
// Create directory
dirPath := filepath.Join(targetDir, file.Path)
if err := os.MkdirAll(dirPath, 0755); err != nil {
return fmt.Errorf("failed to create directory %s: %w", dirPath, err)
}
continue
}
// Check if we can skip downloading this file
if localParentFiles != nil {
if parentFile, exists := localParentFiles[file.Path]; exists && parentFile.SHA256 == file.SHA256 {
// File exists in parent with same hash, copy it instead of downloading
parentFilePath := filepath.Join(targetDir, "..", localParentID, file.Path)
targetFilePath := filepath.Join(targetDir, file.Path)
// Ensure parent directory exists
if err := os.MkdirAll(filepath.Dir(targetFilePath), 0755); err != nil {
return fmt.Errorf("failed to create directory for %s: %w", targetFilePath, err)
}
// Copy the file
if err := copyFile(parentFilePath, targetFilePath); err != nil {
// If copy fails, fall back to downloading
fmt.Printf("Failed to copy file %s, will download instead: %v\n", file.Path, err)
} else {
// Skip to next file
continue
}
}
}
// Download the file
if err := c.downloadFile(ctx, snapshotID, file.Path, filepath.Join(targetDir, file.Path)); err != nil {
return fmt.Errorf("failed to download file %s: %w", file.Path, err)
}
}
return nil
}
// downloadFile downloads a single file from the server
func (c *Client) downloadFile(ctx context.Context, snapshotID, filePath, targetPath string) error {
// Create the request
req := &agateGrpc.DownloadFileRequest{
// GetDiffInfo gets the hash and size of a differential archive.
func (c *Client) GetDiffInfo(ctx context.Context, snapshotID, localParentID string) (*store.DiffInfo, error) {
req := &agateGrpc.GetDiffInfoRequest{
SnapshotId: snapshotID,
FilePath: filePath,
LocalParentId: localParentID,
}
// Start streaming the file
stream, err := c.client.DownloadFile(ctx, req)
info, err := c.client.GetDiffInfo(ctx, req)
if err != nil {
return fmt.Errorf("failed to start file download: %w", err)
return nil, fmt.Errorf("failed to get diff info: %w", err)
}
return &store.DiffInfo{
SHA256: info.Sha256Hash,
Size: info.SizeBytes,
}, nil
}
// DownloadSnapshotDiff скачивает архив с разницей между снапшотами.
func (c *Client) DownloadSnapshotDiff(ctx context.Context, snapshotID, localParentID, targetPath string) error {
// Check for local file and validate it
if fileInfo, err := os.Stat(targetPath); err == nil {
remoteDiffInfo, err := c.GetDiffInfo(ctx, snapshotID, localParentID)
if err != nil {
// Log the error but proceed with download
fmt.Printf("could not get remote diff info: %v. proceeding with download.", err)
} else {
if fileInfo.Size() == remoteDiffInfo.Size {
localHash, err := hash.CalculateFileHash(targetPath)
if err == nil && localHash == remoteDiffInfo.SHA256 {
fmt.Printf("local snapshot archive %s is valid, skipping download.", targetPath)
return nil // File is valid, skip download
}
}
}
}
var offset int64
fileInfo, err := os.Stat(targetPath)
if err == nil {
offset = fileInfo.Size()
} else if !os.IsNotExist(err) {
return fmt.Errorf("failed to stat temporary file: %w", err)
}
req := &agateGrpc.DownloadSnapshotDiffRequest{
SnapshotId: snapshotID,
LocalParentId: localParentID,
Offset: offset,
}
stream, err := c.client.DownloadSnapshotDiff(ctx, req)
if err != nil {
return fmt.Errorf("failed to start snapshot diff download: %w", err)
}
// Ensure the target directory exists
if err := os.MkdirAll(filepath.Dir(targetPath), 0755); err != nil {
return fmt.Errorf("failed to create directory for %s: %w", targetPath, err)
}
// Create the target file
file, err := os.Create(targetPath)
file, err := os.OpenFile(targetPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
if err != nil {
return fmt.Errorf("failed to create file %s: %w", targetPath, err)
return fmt.Errorf("failed to open file %s: %w", targetPath, err)
}
defer file.Close()
defer func() {
if err := file.Close(); err != nil {
fmt.Printf("failed to close file: %v", err)
}
}()
// Receive and write chunks
for {
resp, err := stream.Recv()
if err == io.EOF {
break
}
if err != nil {
return fmt.Errorf("error receiving file chunk: %w", err)
return fmt.Errorf("error receiving diff chunk: %w", err)
}
// Write the chunk to the file
if _, err := file.Write(resp.ChunkData); err != nil {
return fmt.Errorf("error writing to file: %w", err)
}
@@ -217,26 +175,3 @@ func (c *Client) downloadFile(ctx context.Context, snapshotID, filePath, targetP
return nil
}
// Helper function to copy a file
func copyFile(src, dst string) error {
sourceFile, err := os.Open(src)
if err != nil {
return err
}
defer sourceFile.Close()
destFile, err := os.Create(dst)
if err != nil {
return err
}
defer destFile.Close()
_, err = io.Copy(destFile, sourceFile)
return err
}
// Connect creates a new client connected to the specified address
func Connect(address string) (*Client, error) {
return NewClient(address)
}

View File

@@ -14,21 +14,21 @@ import (
"gitea.unprism.ru/KRBL/Agate/store"
)
// Server implements the gRPC server for snapshots
// Server реализует gRPC-сервер для снапшотов.
type Server struct {
agateGrpc.UnimplementedSnapshotServiceServer
manager interfaces.SnapshotManager
server *stdgrpc.Server
}
// NewServer creates a new snapshot server
// NewServer создает новый сервер снапшотов.
func NewServer(manager interfaces.SnapshotManager) *Server {
return &Server{
manager: manager,
}
}
// Start starts the gRPC server on the specified address
// Start запускает gRPC-сервер на указанном адресе.
func (s *Server) Start(ctx context.Context, address string) error {
lis, err := net.Listen("tcp", address)
if err != nil {
@@ -45,23 +45,24 @@ func (s *Server) Start(ctx context.Context, address string) error {
}()
fmt.Printf("Server started on %s\n", address)
// Ждем отмены контекста для остановки сервера
<-ctx.Done()
s.Stop()
return nil
}
// Stop gracefully stops the server
func (s *Server) Stop(ctx context.Context) error {
// Stop изящно останавливает сервер.
func (s *Server) Stop() {
if s.server != nil {
s.server.GracefulStop()
fmt.Println("Server stopped")
}
return nil
}
// ListSnapshots implements the gRPC ListSnapshots method
// ListSnapshots реализует gRPC-метод ListSnapshots.
func (s *Server) ListSnapshots(ctx context.Context, req *agateGrpc.ListSnapshotsRequest) (*agateGrpc.ListSnapshotsResponse, error) {
// Create empty ListOptions since the proto doesn't have active filter/pagination fields yet
opts := store.ListOptions{}
snapshots, err := s.manager.ListSnapshots(ctx, opts)
if err != nil {
return nil, fmt.Errorf("failed to list snapshots: %w", err)
@@ -78,7 +79,7 @@ func (s *Server) ListSnapshots(ctx context.Context, req *agateGrpc.ListSnapshots
return response, nil
}
// GetSnapshotDetails implements the gRPC GetSnapshotDetails method
// GetSnapshotDetails реализует gRPC-метод GetSnapshotDetails.
func (s *Server) GetSnapshotDetails(ctx context.Context, req *agateGrpc.GetSnapshotDetailsRequest) (*agateGrpc.SnapshotDetails, error) {
snapshot, err := s.manager.GetSnapshotDetails(ctx, req.SnapshotId)
if err != nil {
@@ -107,17 +108,15 @@ func (s *Server) GetSnapshotDetails(ctx context.Context, req *agateGrpc.GetSnaps
return response, nil
}
// DownloadFile implements the gRPC DownloadFile method
// DownloadFile реализует gRPC-метод DownloadFile.
func (s *Server) DownloadFile(req *agateGrpc.DownloadFileRequest, stream agateGrpc.SnapshotService_DownloadFileServer) error {
// Open the file from the snapshot
fileReader, err := s.manager.OpenFile(context.Background(), req.SnapshotId, req.FilePath)
if err != nil {
return fmt.Errorf("failed to open file: %w", err)
}
defer fileReader.Close()
// Read the file in chunks and send them to the client
buffer := make([]byte, 64*1024) // 64KB chunks
buffer := make([]byte, 64*1024)
for {
n, err := fileReader.Read(buffer)
if err == io.EOF {
@@ -126,19 +125,53 @@ func (s *Server) DownloadFile(req *agateGrpc.DownloadFileRequest, stream agateGr
if err != nil {
return fmt.Errorf("failed to read file: %w", err)
}
// Send the chunk to the client
if err := stream.Send(&agateGrpc.DownloadFileResponse{
ChunkData: buffer[:n],
}); err != nil {
if err := stream.Send(&agateGrpc.DownloadFileResponse{ChunkData: buffer[:n]}); err != nil {
return fmt.Errorf("failed to send chunk: %w", err)
}
}
return nil
}
// Helper function to convert store.SnapshotInfo to grpc.SnapshotInfo
// DownloadSnapshotDiff реализует gRPC-метод DownloadSnapshotDiff.
func (s *Server) DownloadSnapshotDiff(req *agateGrpc.DownloadSnapshotDiffRequest, stream agateGrpc.SnapshotService_DownloadSnapshotDiffServer) error {
diffReader, err := s.manager.StreamSnapshotDiff(context.Background(), req.SnapshotId, req.LocalParentId, req.Offset)
if err != nil {
return fmt.Errorf("failed to stream snapshot diff: %w", err)
}
defer diffReader.Close()
buffer := make([]byte, 64*1024)
for {
n, err := diffReader.Read(buffer)
if err == io.EOF {
break
}
if err != nil {
return fmt.Errorf("failed to read from diff stream: %w", err)
}
if n > 0 {
if err := stream.Send(&agateGrpc.DownloadFileResponse{ChunkData: buffer[:n]}); err != nil {
return fmt.Errorf("failed to send diff chunk: %w", err)
}
}
}
return nil
}
// GetDiffInfo реализует gRPC-метод GetDiffInfo.
func (s *Server) GetDiffInfo(ctx context.Context, req *agateGrpc.GetDiffInfoRequest) (*agateGrpc.DiffInfo, error) {
diffInfo, err := s.manager.GetSnapshotDiffInfo(ctx, req.SnapshotId, req.LocalParentId)
if err != nil {
return nil, fmt.Errorf("failed to get diff info: %w", err)
}
return &agateGrpc.DiffInfo{
Sha256Hash: diffInfo.SHA256,
SizeBytes: diffInfo.Size,
}, nil
}
// Вспомогательная функция для конвертации store.SnapshotInfo в grpc.SnapshotInfo
func convertToGrpcSnapshotInfo(info store.SnapshotInfo) *agateGrpc.SnapshotInfo {
return &agateGrpc.SnapshotInfo{
Id: info.ID,
@@ -147,12 +180,3 @@ func convertToGrpcSnapshotInfo(info store.SnapshotInfo) *agateGrpc.SnapshotInfo
CreationTime: timestamppb.New(info.CreationTime),
}
}
// RunServer is a helper function to create and start a snapshot server
func RunServer(ctx context.Context, manager interfaces.SnapshotManager, address string) (*Server, error) {
server := NewServer(manager)
if err := server.Start(ctx, address); err != nil {
return nil, err
}
return server, nil
}

View File

@@ -1,9 +0,0 @@
package store
import "errors"
// Common errors that can be used by store implementations
var (
// ErrNotFound means that a requested resource was not found
ErrNotFound = errors.New("resource not found")
)

View File

@@ -7,6 +7,7 @@ import (
"os"
"path/filepath"
"gitea.unprism.ru/KRBL/Agate/models"
"gitea.unprism.ru/KRBL/Agate/store"
)
@@ -75,7 +76,7 @@ func (fs *fileSystemStore) RetrieveBlob(ctx context.Context, snapshotID string)
if err != nil {
if os.IsNotExist(err) {
// Если файл не найден, возвращаем кастомную ошибку
return nil, store.ErrNotFound
return nil, models.ErrNotFound
}
return nil, fmt.Errorf("failed to open blob file %s: %w", blobPath, err)
}
@@ -109,7 +110,7 @@ func (fs *fileSystemStore) GetBlobPath(ctx context.Context, snapshotID string) (
// Проверяем существование файла
if _, err := os.Stat(blobPath); err != nil {
if os.IsNotExist(err) {
return "", store.ErrNotFound
return "", models.ErrNotFound
}
return "", fmt.Errorf("failed to stat blob file %s: %w", blobPath, err)
}

View File

@@ -6,11 +6,13 @@ import (
"encoding/json"
"errors"
"fmt"
"gitea.unprism.ru/KRBL/Agate/store"
_ "github.com/mattn/go-sqlite3"
"os"
"path/filepath"
"time"
"gitea.unprism.ru/KRBL/Agate/models"
"gitea.unprism.ru/KRBL/Agate/store"
_ "github.com/mattn/go-sqlite3"
)
const (
@@ -130,7 +132,7 @@ func (s *sqliteStore) GetSnapshotMetadata(ctx context.Context, snapshotID string
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
// Если запись не найдена, возвращаем кастомную ошибку
return nil, store.ErrNotFound
return nil, models.ErrNotFound
}
return nil, fmt.Errorf("failed to query snapshot %s: %w", snapshotID, err)
}

View File

@@ -6,6 +6,12 @@ import (
"time"
)
// DiffInfo represents metadata about a differential archive.
type DiffInfo struct {
SHA256 string
Size int64
}
// FileInfo represents metadata and attributes of a file or directory.
type FileInfo struct {
Path string // Path represents the relative or absolute location of the file or directory in the filesystem.
@@ -31,6 +37,14 @@ type SnapshotInfo struct {
CreationTime time.Time // Время создания
}
// SnapshotStatus represents the status of an asynchronous snapshot creation process.
type SnapshotStatus struct {
ID string // Unique identifier of the job (usually same as Snapshot ID)
Status string // Current status: "pending", "running", "done", "failed"
Progress float64 // Progress from 0.0 to 1.0
Error string // Error message if failed
}
// ListOptions provides options for filtering and paginating snapshot lists
type ListOptions struct {
FilterByName string // Filter snapshots by name (substring match)