fix(storagenode): support partial success/failure for append

ijsong · ijsong · commit 5d438bb111cb · 2023-06-04T16:36:56.000+09:00
The Append RPC function allows batch logging but may occasionally only partially succeed. The
storage node previously overlooked partial success, but this PR fixes to consider partial success.
diff --git a/internal/storagenode/logstream/append.go b/internal/storagenode/logstream/append.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"time"
 
+	"go.uber.org/zap"
+
 	"github.com/kakao/varlog/internal/batchlet"
 	snerrors "github.com/kakao/varlog/internal/storagenode/errors"
 	"github.com/kakao/varlog/pkg/verrors"
@@ -155,16 +157,24 @@ func (lse *Executor) waitForCompletionOfAppends(ctx context.Context, dataBatchLe
 	result := make([]snpb.AppendResult, dataBatchLen)
 	for i := range awgs {
 		cerr := awgs[i].wait(ctx)
-		if err == nil && cerr != nil {
-			err = cerr
+		if cerr != nil {
+			result[i].Error = cerr.Error()
+			if err == nil {
+				err = cerr
+			}
+			continue
 		}
-		if cerr == nil {
-			result[i].Meta.TopicID = lse.tpid
-			result[i].Meta.LogStreamID = lse.lsid
-			result[i].Meta.GLSN = awgs[i].glsn
-			result[i].Meta.LLSN = awgs[i].llsn
-			awgs[i].release()
+		if err != nil {
+			lse.logger.Panic("Results of batch requests of Append RPC must not be interleaved with success and failure", zap.Error(err))
 		}
+		result[i].Meta.TopicID = lse.tpid
+		result[i].Meta.LogStreamID = lse.lsid
+		result[i].Meta.GLSN = awgs[i].glsn
+		result[i].Meta.LLSN = awgs[i].llsn
+		awgs[i].release()
+	}
+	if result[0].Meta.GLSN.Invalid() {
+		return nil, err
 	}
-	return result, err
+	return result, nil
 }
diff --git a/internal/storagenode/storagenode_test.go b/internal/storagenode/storagenode_test.go
@@ -601,6 +601,67 @@ func TestStorageNode_Append(t *testing.T) {
 				}()
 			},
 		},
+		{
+			name: "AppendBatch",
+			testf: func(t *testing.T, addr string, lc *client.LogClient) {
+				lss, lastGLSN := TestSealLogStreamReplica(t, cid, snid, tpid, lsid, types.InvalidGLSN, addr)
+				require.Equal(t, varlogpb.LogStreamStatusSealed, lss)
+				require.True(t, lastGLSN.Invalid())
+
+				TestUnsealLogStreamReplica(t, cid, snid, tpid, lsid, []varlogpb.LogStreamReplica{
+					{
+						StorageNode: varlogpb.StorageNode{
+							StorageNodeID: snid,
+							Address:       addr,
+						},
+						TopicLogStream: varlogpb.TopicLogStream{
+							TopicID:     tpid,
+							LogStreamID: lsid,
+						},
+					},
+				}, addr)
+
+				batch := [][]byte{[]byte("msg1"), []byte("msg2"), []byte("msg3")}
+				var wg sync.WaitGroup
+				wg.Add(1)
+				go func() {
+					defer wg.Done()
+					res, err := lc.Append(context.Background(), tpid, lsid, batch)
+					require.NoError(t, err)
+					require.Len(t, res, len(batch))
+					require.Empty(t, res[0].Error)
+					require.False(t, res[0].Meta.GLSN.Invalid())
+					require.NotEmpty(t, res[1].Error)
+					require.True(t, res[1].Meta.GLSN.Invalid())
+					require.NotEmpty(t, res[2].Error)
+					require.True(t, res[2].Meta.GLSN.Invalid())
+				}()
+
+				require.Eventually(t, func() bool {
+					reportcommitter.TestCommit(t, addr, snpb.CommitRequest{
+						StorageNodeID: snid,
+						CommitResult: snpb.LogStreamCommitResult{
+							TopicID:             tpid,
+							LogStreamID:         lsid,
+							CommittedLLSNOffset: 1,
+							CommittedGLSNOffset: 1,
+							CommittedGLSNLength: 1,
+							Version:             1,
+							HighWatermark:       1,
+						},
+					})
+					reports := reportcommitter.TestGetReport(t, addr)
+					require.Len(t, reports, 1)
+					return reports[0].Version == types.Version(1)
+				}, time.Second, 10*time.Millisecond)
+
+				lss, lastGLSN = TestSealLogStreamReplica(t, cid, snid, tpid, lsid, 1, addr)
+				require.Equal(t, varlogpb.LogStreamStatusSealed, lss)
+				require.Equal(t, types.GLSN(1), lastGLSN)
+
+				wg.Wait()
+			},
+		},
 	}
 
 	for _, tc := range tcs {
