FMWK-816 Optimize S3 operations

backup_restore_test.go

@@ -1198,13 +1198,13 @@ func TestRestoreExpiredRecords(t *testing.T) {
 		options.WithDir(directory),
 	)
 	require.NoError(t, err)
-	w, err := writer.NewWriter(ctx, fmt.Sprintf("%s-%s.asb", testASNamespace, setName))
+	w, err := writer.NewWriter(ctx, fmt.Sprintf("%s-%s.asb", testASNamespace, setName), true)
 	require.NoError(t, err)
 	require.NotNil(t, w)
 
 	encoder := NewEncoder[*models.Token](EncoderTypeASB, testASNamespace, false, false)
 
-	header := encoder.GetHeader(0, false)
+	header := encoder.GetHeader(0, true)
 
 	_, err = w.Write(header)
 	require.NoError(t, err)

handler_backup.go

@@ -42,7 +42,8 @@ type Writer interface {
 	// a new writer, they might be working in parallel. Backup logic will close
 	// the writer after backup is done. Header func is executed on a writer
 	// after creation (on each one in case of multipart file).
-	NewWriter(ctx context.Context, filename string) (io.WriteCloser, error)
+	// isRecords indicates whether the file contains records data.
+	NewWriter(ctx context.Context, filename string, isRecords bool) (io.WriteCloser, error)
 
 	// GetType returns the type of storage. Used in logging.
 	GetType() string

io/encoding/asb/encode.go

@@ -103,11 +103,11 @@ func (e *Encoder[T]) encodeSIndex(sIndex *models.SIndex, buff *bytes.Buffer) (in
 	return sindexToASB(sIndex, buff)
 }
 
-func (e *Encoder[T]) GetHeader(_ uint64, isMeta bool) []byte {
+func (e *Encoder[T]) GetHeader(_ uint64, isRecords bool) []byte {
 	// capacity is arbitrary, just probably enough to avoid reallocations
 	buff := bytes.NewBuffer(make([]byte, 0, 1024))
 
-	writeVersionText(e.headerVersion(isMeta), buff)
+	writeVersionText(e.headerVersion(isRecords), buff)
 
 	writeNamespaceMetaText(e.config.Namespace, buff)
 
@@ -118,12 +118,12 @@ func (e *Encoder[T]) GetHeader(_ uint64, isMeta bool) []byte {
 	return buff.Bytes()
 }
 
-func (e *Encoder[T]) headerVersion(isMeta bool) string {
-	if isMeta {
-		return e.config.getVersion().toString()
+func (e *Encoder[T]) headerVersion(isRecords bool) string {
+	if isRecords {
+		return version31.toString()
 	}
 
-	return version31.toString()
+	return e.config.getVersion().toString()
 }
 
 // **** META DATA ****

io/encoding/asb/encode_test.go

@@ -190,11 +190,11 @@ func TestGetHeaderFirst(t *testing.T) {
 	expected := "Version 3.1\n# namespace test\n# first-file\n"
 
 	encoder := NewEncoder[*models.Token](testEncoderConfig)
-	firstHeader := encoder.GetHeader(0, false)
+	firstHeader := encoder.GetHeader(0, true)
 	require.Equal(t, expected, string(firstHeader))
 
 	secondExpected := "Version 3.1\n# namespace test\n"
-	secondHeader := encoder.GetHeader(0, false)
+	secondHeader := encoder.GetHeader(0, true)
 	require.Equal(t, secondExpected, string(secondHeader))
 }
 

io/storage/aws/s3/range_reader.go

@@ -71,6 +71,9 @@ func newRangeReader(ctx context.Context, client s3Getter, bucket, key *string) (
 
 // OpenRange opens a file by range.
 func (r *rangeReader) OpenRange(ctx context.Context, offset, count int64) (io.ReadCloser, error) {
+	// We can't validate checksum for range requests, so we don't set ChecksumMode param in GetObjectInput.
+	// Checksums are generated on upload by S3 for chunk, so when we request data by range, we can't validate its checksum.
+	// Link to issue: https://github.com/aws/aws-sdk-java-v2/issues/5421
 	resp, err := r.client.GetObject(ctx, &s3.GetObjectInput{
 		Bucket:  r.bucket,
 		Key:     r.key,

io/storage/aws/s3/reader_test.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"path"
 	"path/filepath"
 	"strings"
 	"sync"
@@ -311,8 +312,8 @@ func (s *AwsSuite) TestReader_StreamPathList() {
 	})
 
 	pathList := []string{
-		filepath.Join(testFolderPathList, "1732519390025"),
-		filepath.Join(testFolderPathList, "1732519590025"),
+		path.Join(testFolderPathList, "1732519390025"),
+		path.Join(testFolderPathList, "1732519590025"),
 	}
 
 	reader, err := NewReader(
@@ -362,8 +363,8 @@ func (s *AwsSuite) TestReader_StreamFilesList() {
 	})
 
 	pathList := []string{
-		filepath.Join(testFolderFileList, "backup_1.asb"),
-		filepath.Join(testFolderFileList, "backup_2.asb"),
+		path.Join(testFolderFileList, "backup_1.asb"),
+		path.Join(testFolderFileList, "backup_2.asb"),
 	}
 
 	reader, err := NewReader(
@@ -1002,8 +1003,8 @@ func (s *AwsSuite) TestReader_SetObjectsToStream() {
 
 	// Define a list of objects to stream
 	objectsToStream := []string{
-		filepath.Join(testFolderWithData, fmt.Sprintf(testFileNameAsbTemplate, 0)),
-		filepath.Join(testFolderWithData, fmt.Sprintf(testFileNameAsbTemplate, 1)),
+		path.Join(testFolderWithData, fmt.Sprintf(testFileNameAsbTemplate, 0)),
+		path.Join(testFolderWithData, fmt.Sprintf(testFileNameAsbTemplate, 1)),
 	}
 
 	// Set the objects to stream