KRBL/Agate
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Добавлена локальная загрузка снапшота

Browse Source
This commit is contained in:
Александр Лазаренко
2025-10-30 00:20:28 +03:00
parent 5192658607
commit f34539c06b
10 changed files with 326 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
Makefile
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

168
grpc/snapshot.pb.go
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
@@ -500,6 +500,112 @@ func (x *DownloadSnapshotDiffRequest) GetOffset() int64 {
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 = "" +
@@ -538,12 +644,22 @@ const file_snapshot_proto_rawDesc = "" +
"\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\x06offset2\xf1\x03\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\x01\x12e\n" +
"\x14DownloadSnapshotDiff\x12'.agate.grpc.DownloadSnapshotDiffRequest\x1a .agate.grpc.DownloadFileResponse\"\x000\x01B\"Z gitea.unprism.ru/KRBL/Agate/grpcb\x06proto3"
"\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
@@ -557,7 +673,7 @@ func file_snapshot_proto_rawDescGZIP() []byte {
return file_snapshot_proto_rawDescData
}
var file_snapshot_proto_msgTypes = make([]protoimpl.MessageInfo, 9)
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
@@ -568,26 +684,30 @@ var file_snapshot_proto_goTypes = []any{
(*DownloadFileRequest)(nil), // 6: agate.grpc.DownloadFileRequest
(*DownloadFileResponse)(nil), // 7: agate.grpc.DownloadFileResponse
(*DownloadSnapshotDiffRequest)(nil), // 8: agate.grpc.DownloadSnapshotDiffRequest
(*timestamppb.Timestamp)(nil), // 9: google.protobuf.Timestamp
(*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{
9, // 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
8, // 7: agate.grpc.SnapshotService.DownloadSnapshotDiff:input_type -> agate.grpc.DownloadSnapshotDiffRequest
4, // 8: agate.grpc.SnapshotService.ListSnapshots:output_type -> agate.grpc.ListSnapshotsResponse
2, // 9: agate.grpc.SnapshotService.GetSnapshotDetails:output_type -> agate.grpc.SnapshotDetails
7, // 10: agate.grpc.SnapshotService.DownloadFile:output_type -> agate.grpc.DownloadFileResponse
7, // 11: agate.grpc.SnapshotService.DownloadSnapshotDiff:output_type -> agate.grpc.DownloadFileResponse
8, // [8:12] is the sub-list for method output_type
4, // [4:8] 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
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
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
}
func init() { file_snapshot_proto_init() }
@@ -601,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: 9,
NumMessages: 11,
NumExtensions: 0,
NumServices: 1,
},

12
grpc/snapshot.pb.gw.go
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")

15
grpc/snapshot.proto
View File

@@ -32,6 +32,9 @@ service SnapshotService {
// Скачать архив, содержащий только разницу между двумя снапшотами
rpc DownloadSnapshotDiff(DownloadSnapshotDiffRequest) returns (stream DownloadFileResponse) {}
// Получить информацию о дифе (хеш и размер)
rpc GetDiffInfo(GetDiffInfoRequest) returns (DiffInfo) {}
}
// Метаданные файла внутри снапшота
@@ -86,3 +89,15 @@ message DownloadSnapshotDiffRequest {
string local_parent_id = 2; // ID снапшота, который уже есть у клиента
int64 offset = 3; // Смещение в байтах для докачки
}
// Запрос на получение информации о дифе
message GetDiffInfoRequest {
string snapshot_id = 1;
string local_parent_id = 2;
}
// Информация о дифе
message DiffInfo {
string sha256_hash = 1;
int64 size_bytes = 2;
}

42
grpc/snapshot_grpc.pb.go
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
@@ -23,6 +23,7 @@ const (
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.
@@ -39,6 +40,8 @@ type SnapshotServiceClient interface {
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 {
@@ -107,6 +110,16 @@ func (c *snapshotServiceClient) DownloadSnapshotDiff(ctx context.Context, in *Do
// 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.
@@ -121,6 +134,8 @@ type SnapshotServiceServer interface {
DownloadFile(*DownloadFileRequest, grpc.ServerStreamingServer[DownloadFileResponse]) error
// Скачать архив, содержащий только разницу между двумя снапшотами
DownloadSnapshotDiff(*DownloadSnapshotDiffRequest, grpc.ServerStreamingServer[DownloadFileResponse]) error
// Получить информацию о дифе (хеш и размер)
GetDiffInfo(context.Context, *GetDiffInfoRequest) (*DiffInfo, error)
mustEmbedUnimplementedSnapshotServiceServer()
}
@@ -143,6 +158,9 @@ func (UnimplementedSnapshotServiceServer) DownloadFile(*DownloadFileRequest, grp
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() {}
@@ -222,6 +240,24 @@ func _SnapshotService_DownloadSnapshotDiff_Handler(srv interface{}, stream grpc.
// 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)
@@ -237,6 +273,10 @@ var SnapshotService_ServiceDesc = grpc.ServiceDesc{
MethodName: "GetSnapshotDetails",
Handler: _SnapshotService_GetSnapshotDetails_Handler,
},
{
MethodName: "GetDiffInfo",
Handler: _SnapshotService_GetDiffInfo_Handler,
},
},
Streams: []grpc.StreamDesc{
{

3
interfaces/snapshot.go
View File

@@ -38,6 +38,9 @@ type SnapshotManager interface {
// 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

70
manager.go
View File

@@ -403,6 +403,37 @@ func (data *SnapshotManagerData) UpdateSnapshotMetadata(ctx context.Context, sna
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
@@ -418,10 +449,10 @@ func (r *diffArchiveReader) Close() error {
return err
}
func (data *SnapshotManagerData) StreamSnapshotDiff(ctx context.Context, snapshotID, parentID string, offset int64) (io.ReadCloser, error) {
func (data *SnapshotManagerData) createDiffArchive(ctx context.Context, snapshotID, parentID string) (string, string, error) {
targetSnap, err := data.metadataStore.GetSnapshotMetadata(ctx, snapshotID)
if err != nil {
return nil, fmt.Errorf("failed to get target snapshot metadata: %w", err)
return "", "", fmt.Errorf("failed to get target snapshot metadata: %w", err)
}
parentFiles := make(map[string]string)
@@ -449,38 +480,38 @@ func (data *SnapshotManagerData) StreamSnapshotDiff(ctx context.Context, snapsho
}
if len(filesToInclude) == 0 {
return io.NopCloser(bytes.NewReader(nil)), nil
return "", "", nil
}
tempStagingDir, err := os.MkdirTemp(data.blobStore.GetBaseDir(), "diff-staging-*")
if err != nil {
return nil, fmt.Errorf("failed to create temp staging directory: %w", err)
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 nil, err
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 nil, fmt.Errorf("failed to create dir for diff file: %w", err)
return "", "", fmt.Errorf("failed to create dir for diff file: %w", err)
}
fileWriter, err := os.Create(destPath)
if err != nil {
os.RemoveAll(tempStagingDir)
return nil, err
return "", "", err
}
err = archive.ExtractFileFromArchive(targetBlobPath, filePath, fileWriter)
fileWriter.Close()
if err != nil {
os.RemoveAll(tempStagingDir)
return nil, fmt.Errorf("failed to extract file %s for diff: %w", filePath, err)
return "", "", fmt.Errorf("failed to extract file %s for diff: %w", filePath, err)
}
}
@@ -488,12 +519,27 @@ func (data *SnapshotManagerData) StreamSnapshotDiff(ctx context.Context, snapsho
if err := archive.CreateArchive(tempStagingDir, tempArchivePath); err != nil {
os.RemoveAll(tempStagingDir)
os.Remove(tempArchivePath)
return nil, fmt.Errorf("failed to create diff archive: %w", err)
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 {
os.RemoveAll(tempStagingDir)
if tempStagingDir != "" {
os.RemoveAll(tempStagingDir)
}
os.Remove(tempArchivePath)
return nil, err
}
@@ -501,7 +547,9 @@ func (data *SnapshotManagerData) StreamSnapshotDiff(ctx context.Context, snapsho
if offset > 0 {
if _, err := archiveFile.Seek(offset, io.SeekStart); err != nil {
archiveFile.Close()
os.RemoveAll(tempStagingDir)
if tempStagingDir != "" {
os.RemoveAll(tempStagingDir)
}
os.Remove(tempArchivePath)
return nil, fmt.Errorf("failed to seek in diff archive: %w", err)
}

36
remote/client.go
View File

@@ -11,6 +11,7 @@ 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"
)
@@ -89,8 +90,43 @@ func (c *Client) FetchSnapshotDetails(ctx context.Context, snapshotID string) (*
return snapshot, nil
}
// 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,
LocalParentId: localParentID,
}
info, err := c.client.GetDiffInfo(ctx, req)
if err != nil {
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 {

13
remote/server.go
View File

@@ -158,6 +158,19 @@ func (s *Server) DownloadSnapshotDiff(req *agateGrpc.DownloadSnapshotDiffRequest
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{

6
store/store.go
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.