incremental object aware write at

Signed-off-by: Lyndon-Li <lyonghui@vmware.com>

changelogs/unreleased/9887-Lyndon-Li

@@ -0,0 +1 @@
+Add WriteAt implementation for Incremental aware object writer for block data mover

pkg/repository/udmrepo/kopialib/lib_repo.go

@@ -92,7 +92,6 @@ type kopiaObjectWriterEx struct {
 	description      string
 	compressor       compression.Name
 	splitter         string
-	zeroBuffer       []byte
 	zeroObject       object.ID
 	writeLock        sync.Mutex
 	asyncWritesSem   chan struct{}
@@ -1045,6 +1044,11 @@ func (kow *kopiaObjectWriterEx) WriteAt(p []byte, offset int64) (int, error) {
 
 			for i := startEntry; i < endEntry; i++ {
 				e := kow.parentEntries[i]
+
+				if e.Start != int64(i)*kow.blockSize {
+					return 0, errors.Errorf("parent entry %v start %v does not match expected start %v", i, e.Start, int64(i)*kow.blockSize)
+				}
+
 				if e.Length != kow.blockSize {
 					return 0, errors.Errorf("parent entry %v length %v does not match child block size %v", i, e.Length, kow.blockSize)
 				}
@@ -1069,7 +1073,7 @@ func (kow *kopiaObjectWriterEx) WriteAt(p []byte, offset int64) (int, error) {
 
 		objName := fmt.Sprintf("%s-b%v", kow.description, entryID)
 		if err := kow.writeZeroObject(objName, entryID); err != nil {
-			return 0, errors.Wrapf(err, "error writting zero object for %s", objName)
+			return 0, errors.Wrapf(err, "error writing zero object for %s", objName)
 		}
 
 		curPos += kow.blockSize

pkg/repository/udmrepo/kopialib/lib_repo_ex_test.go

@@ -258,6 +258,7 @@ func TestKopiaObjectWriterEx_Write(t *testing.T) {
 		{
 			name: "write object returns nil writer",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				mockRepoWriter.On("NewObjectWriter", mock.Anything, mock.Anything).Return(nil)
 
@@ -271,14 +272,16 @@ func TestKopiaObjectWriterEx_Write(t *testing.T) {
 			inputData:   make([]byte, 1024),
 			expectedLen: 1024,
 			verify: func(t *testing.T, kow *kopiaObjectWriterEx) {
+				t.Helper()
 				err := kow.getWriteError()
-				assert.Error(t, err)
-				assert.Contains(t, err.Error(), "error openning writer for -b0")
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), "error opening writer for -b0")
 			},
 		},
 		{
 			name: "write object result error",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				mockWriter := repomocks.NewWriter(t)
 
@@ -299,8 +302,9 @@ func TestKopiaObjectWriterEx_Write(t *testing.T) {
 			inputData:   make([]byte, 1024),
 			expectedLen: 1024,
 			verify: func(t *testing.T, kow *kopiaObjectWriterEx) {
+				t.Helper()
 				err := kow.getWriteError()
-				assert.Error(t, err)
+				require.Error(t, err)
 				assert.Contains(t, err.Error(), "simulated result error")
 			},
 		},
@@ -462,6 +466,7 @@ func TestKopiaObjectWriterEx_Result(t *testing.T) {
 		{
 			name: "write indirect object encoding failure",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				mockWriter := repomocks.NewWriter(t)
 
@@ -481,6 +486,7 @@ func TestKopiaObjectWriterEx_Result(t *testing.T) {
 		{
 			name: "write indirect object result failure",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				mockWriter := repomocks.NewWriter(t)
 
@@ -652,6 +658,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 		{
 			name: "writer is closed",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				return &kopiaObjectWriterEx{
 					rawRepoWriter: nil,
 				}
@@ -663,6 +670,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 		{
 			name: "invalid offset",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				return &kopiaObjectWriterEx{
 					rawRepoWriter: repomocks.NewMockRepositoryWriter(t),
 					blockSize:     1024,
@@ -675,6 +683,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 		{
 			name: "invalid length",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				return &kopiaObjectWriterEx{
 					rawRepoWriter: repomocks.NewMockRepositoryWriter(t),
 					blockSize:     1024,
@@ -687,6 +696,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 		{
 			name: "cannot write back",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				return &kopiaObjectWriterEx{
 					rawRepoWriter: repomocks.NewMockRepositoryWriter(t),
 					blockSize:     1024,
@@ -702,6 +712,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 		{
 			name: "success write at cur pos",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				mockWriter := repomocks.NewWriter(t)
 
@@ -724,13 +735,15 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 			offset:      0,
 			expectedLen: 1024,
 			verify: func(t *testing.T, kow *kopiaObjectWriterEx) {
-				assert.Equal(t, 1, len(kow.entries))
+				t.Helper()
+				assert.Len(t, kow.entries, 1)
 				assert.Equal(t, int64(0), kow.entries[0].Start)
 			},
 		},
 		{
 			name: "success write with gap filling zeros",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				mockWriter := repomocks.NewWriter(t)
 
@@ -754,7 +767,8 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 			offset:      1024,
 			expectedLen: 1024,
 			verify: func(t *testing.T, kow *kopiaObjectWriterEx) {
-				assert.Equal(t, 2, len(kow.entries))
+				t.Helper()
+				assert.Len(t, kow.entries, 2)
 				assert.Equal(t, int64(0), kow.entries[0].Start)
 				id, _ := object.ParseID("I12345")
 				assert.Equal(t, id, kow.entries[0].Object)
@@ -765,6 +779,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 		{
 			name: "success write with gap filling from parent",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				mockWriter := repomocks.NewWriter(t)
 
@@ -791,7 +806,8 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 			offset:      1024,
 			expectedLen: 1024,
 			verify: func(t *testing.T, kow *kopiaObjectWriterEx) {
-				assert.Equal(t, 2, len(kow.entries))
+				t.Helper()
+				assert.Len(t, kow.entries, 2)
 				assert.Equal(t, int64(0), kow.entries[0].Start)
 				parentID, _ := object.ParseID("Iparent")
 				assert.Equal(t, parentID, kow.entries[0].Object)
@@ -801,6 +817,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 		{
 			name: "success write zero length",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				return &kopiaObjectWriterEx{
 					ctx:           context.Background(),
@@ -813,12 +830,14 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 			offset:      0,
 			expectedLen: 0,
 			verify: func(t *testing.T, kow *kopiaObjectWriterEx) {
-				assert.Equal(t, 0, len(kow.entries))
+				t.Helper()
+				assert.Empty(t, kow.entries)
 			},
 		},
 		{
 			name: "gap filling with invalid parent entry length",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				return &kopiaObjectWriterEx{
 					ctx:           context.Background(),
@@ -837,6 +856,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 		{
 			name: "gap filling partially with parent and rest with zeros",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				mockWriter := repomocks.NewWriter(t)
 
@@ -864,7 +884,8 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 			offset:      2048,
 			expectedLen: 1024,
 			verify: func(t *testing.T, kow *kopiaObjectWriterEx) {
-				assert.Equal(t, 3, len(kow.entries))
+				t.Helper()
+				assert.Len(t, kow.entries, 3)
 				assert.Equal(t, int64(0), kow.entries[0].Start)
 
 				parentID, _ := object.ParseID("Iparent")
@@ -880,6 +901,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 		{
 			name: "writeZeroObject failure",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				mockWriter := repomocks.NewWriter(t)
 
@@ -898,11 +920,12 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 			},
 			inputData:   make([]byte, 1024),
 			offset:      1024,
-			expectedErr: "error writting zero object for -b0: error writting for -b0: simulated zero object write error",
+			expectedErr: "error writing zero object for -b0: error writing for -b0: simulated zero object write error",
 		},
 		{
 			name: "writeObject short write",
 			setupWriter: func(t *testing.T) *kopiaObjectWriterEx {
+				t.Helper()
 				mockRepoWriter := repomocks.NewMockRepositoryWriter(t)
 				mockWriter := repomocks.NewWriter(t)
 
@@ -922,8 +945,9 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 			offset:      0,
 			expectedLen: 1024,
 			verify: func(t *testing.T, kow *kopiaObjectWriterEx) {
+				t.Helper()
 				err := kow.getWriteError()
-				assert.Error(t, err)
+				require.Error(t, err)
 				assert.Contains(t, err.Error(), "short write for -b0")
 			},
 		},
@@ -941,7 +965,7 @@ func TestKopiaObjectWriterEx_WriteAt(t *testing.T) {
 			if tc.expectedErr != "" {
 				assert.EqualError(t, err, tc.expectedErr)
 			} else {
-				assert.NoError(t, err)
+				require.NoError(t, err)
 				assert.Equal(t, tc.expectedLen, l)
 				if tc.verify != nil {
 					tc.verify(t, kow)
@@ -972,14 +996,14 @@ func TestKopiaObjectWriterEx_MultipleWriteAt(t *testing.T) {
 	}
 
 	l, err := kow.WriteAt(make([]byte, 1024), 0)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	assert.Equal(t, 1024, l)
 
 	l, err = kow.WriteAt(make([]byte, 1024), 2048)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	assert.Equal(t, 1024, l)
 
-	assert.Equal(t, 3, len(kow.entries))
+	assert.Len(t, kow.entries, 3)
 	assert.Equal(t, int64(0), kow.entries[0].Start)
 	assert.Equal(t, int64(1024), kow.entries[1].Start)
 	assert.Equal(t, id, kow.entries[1].Object)
@@ -1028,7 +1052,7 @@ func TestKopiaObjectWriterEx_ConcurrentWriteAt(t *testing.T) {
 	close(start)
 	wg.Wait()
 
-	assert.Greater(t, len(kow.entries), 0)
+	assert.NotEmpty(t, kow.entries)
 }
 
 type dummyObjectWriter struct {
@@ -1078,11 +1102,11 @@ func TestKopiaObjectWriterEx_LargeSequentialWrite(t *testing.T) {
 
 	for i := 0; i < blocks; i++ {
 		l, err := kow.Write(data)
-		assert.NoError(t, err)
+		require.NoError(t, err)
 		assert.Equal(t, int(blockSize), l)
 	}
 
-	assert.Equal(t, blocks, len(kow.entries))
+	assert.Len(t, kow.entries, blocks)
 	assert.Equal(t, int64(blocks-1)*blockSize, kow.entries[blocks-1].Start)
 }
 
@@ -1103,11 +1127,11 @@ func TestKopiaObjectWriterEx_LargeSparseWriteAt(t *testing.T) {
 
 	data := make([]byte, blockSize)
 	l, err := kow.WriteAt(data, offset)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	assert.Equal(t, int(blockSize), l)
 
 	expectedEntries := 5121
-	assert.Equal(t, expectedEntries, len(kow.entries))
+	assert.Len(t, kow.entries, expectedEntries)
 	assert.Equal(t, int64(0), kow.entries[0].Start)
 	assert.Equal(t, offset, kow.entries[expectedEntries-1].Start)
 }
@@ -1137,21 +1161,21 @@ func TestKopiaObjectWriterEx_MixedWriteAndWriteAt(t *testing.T) {
 	// 1. Write 1 block sequentially
 	data1 := make([]byte, blockSize)
 	l, err := kow.Write(data1)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	assert.Equal(t, int(blockSize), l)
 
 	// Entries: [0:1024]
-	assert.Equal(t, 1, len(kow.entries))
+	assert.Len(t, kow.entries, 1)
 	assert.Equal(t, int64(0), kow.entries[0].Start)
 
 	// 2. WriteAt with gap (offset = 2048). This creates a gap block at 1024
 	data2 := make([]byte, blockSize)
 	l, err = kow.WriteAt(data2, 2048)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	assert.Equal(t, int(blockSize), l)
 
 	// Entries should now be 3: [0:1024, 1024:2048(zero object), 2048:3072]
-	assert.Equal(t, 3, len(kow.entries))
+	assert.Len(t, kow.entries, 3)
 	assert.Equal(t, int64(0), kow.entries[0].Start)
 	assert.Equal(t, int64(1024), kow.entries[1].Start)
 	assert.Equal(t, id, kow.entries[1].Object) // filled with zero block
@@ -1160,11 +1184,11 @@ func TestKopiaObjectWriterEx_MixedWriteAndWriteAt(t *testing.T) {
 	// 3. Write another block sequentially. It should append at 3072.
 	data3 := make([]byte, blockSize)
 	l, err = kow.Write(data3)
-	assert.NoError(t, err)
+	require.NoError(t, err)
 	assert.Equal(t, int(blockSize), l)
 
 	// Entries should now be 4: [0:1024, 1024:2048(zero object), 2048:3072, 3072:4096]
-	assert.Equal(t, 4, len(kow.entries))
+	assert.Len(t, kow.entries, 4)
 	assert.Equal(t, int64(3072), kow.entries[3].Start)
 }
 
@@ -1194,12 +1218,14 @@ func TestKopiaObjectWriterEx_ConcurrentAsyncErrors(t *testing.T) {
 	// Issue multiple writes so they all spawn async goroutines
 	// First few writes shouldn't fail immediately until getWriteError catches the asynchronous fault
 	for i := 0; i < 10; i++ {
-		kow.Write(data)
+		l, err := kow.Write(data)
+		require.NoError(t, err)
+		assert.Equal(t, 1024, l)
 	}
 
 	id, err := kow.Result()
 
-	assert.Error(t, err)
+	require.Error(t, err)
 	assert.Contains(t, err.Error(), "simulated async error")
 	assert.Equal(t, udmrepo.ID(""), id)
 }
@@ -1232,7 +1258,9 @@ func TestKopiaObjectWriterEx_ConcurrentWriteAndWriteAt(t *testing.T) {
 		go func() {
 			defer wg.Done()
 			<-start
-			kow.Write(make([]byte, 1024))
+			l, err := kow.Write(make([]byte, 1024))
+			require.NoError(t, err)
+			assert.Equal(t, 1024, l)
 		}()
 	}
 
@@ -1255,13 +1283,13 @@ func TestKopiaObjectWriterEx_ConcurrentWriteAndWriteAt(t *testing.T) {
 	wg.Wait()
 
 	// We only care that the locking effectively mitigated a panic or slice data corruption
-	assert.Greater(t, len(kow.entries), 0)
+	assert.NotEmpty(t, kow.entries)
 }
 
 func TestKopiaObjectWriterEx_Checkpoint(t *testing.T) {
 	kow := &kopiaObjectWriterEx{}
 	id, err := kow.Checkpoint()
-	assert.Error(t, err)
+	require.Error(t, err)
 	assert.Equal(t, udmrepo.ID(""), id)
 	assert.Equal(t, "not supported", err.Error())
 }