fix review comments

Signed-off-by: jyjiangkai <[email protected]>

Author: hwjiangkai

client/internal/vanus/store/block_store.go

@@ -28,7 +28,7 @@ import (
 	"google.golang.org/grpc"
 
 	// first-party libraries
-  "github.com/linkall-labs/vanus/client/pkg/codec"
+	"github.com/linkall-labs/vanus/client/pkg/codec"
 	"github.com/linkall-labs/vanus/observability/log"
 	"github.com/linkall-labs/vanus/observability/tracing"
 	"github.com/linkall-labs/vanus/pkg/errors"
@@ -371,15 +371,7 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 		append.releaseStream()
 		// reset new stream connections
 		s.connectAppendStream(ctx)
-		c, _ := append.callbacks.LoadAndDelete(opaqueID)
-		if c != nil {
-			c.(appendCallback)(&segpb.AppendToBlockStreamResponse{
-				Id:           opaqueID,
-				ResponseCode: errpb.ErrorCode_CLOSED,
-				ResponseMsg:  "append stream closed",
-				Offsets:      []int64{},
-			})
-		}
+		append.callbacks.Delete(opaqueID)
 		return nil, err
 	}
 
@@ -394,6 +386,8 @@ func (s *BlockStore) AppendManyStream(ctx context.Context, block uint64, events
 				ResponseMsg:  "append stream context canceled",
 				Offsets:      []int64{},
 			})
+		} else {
+			<-donec
 		}
 	}
 
@@ -489,17 +483,7 @@ func (s *BlockStore) ReadStream(
 		})
 		read.releaseStream()
 		s.connectReadStream(ctx)
-		c, _ := read.callbacks.LoadAndDelete(opaqueID)
-		if c != nil {
-			c.(readCallback)(&segpb.ReadFromBlockStreamResponse{
-				Id:           opaqueID,
-				ResponseCode: errpb.ErrorCode_CLOSED,
-				ResponseMsg:  "read stream closed",
-				Events: &cepb.CloudEventBatch{
-					Events: []*cepb.CloudEvent{},
-				},
-			})
-		}
+		read.callbacks.Delete(opaqueID)
 		return []*ce.Event{}, err
 	}
 
@@ -516,6 +500,8 @@ func (s *BlockStore) ReadStream(
 					Events: []*cepb.CloudEvent{},
 				},
 			})
+		} else {
+			<-donec
 		}
 	}
 

client/pkg/eventlog/eventlog.go

@@ -19,13 +19,12 @@ import (
 	// standard libraries.
 	"context"
 
-	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
-
 	// third-party libraries.
 	ce "github.com/cloudevents/sdk-go/v2"
 
 	// first-party libraries.
 	"github.com/linkall-labs/vanus/client/pkg/api"
+	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
 	segpb "github.com/linkall-labs/vanus/proto/pkg/segment"
 	// this project.
 )

internal/store/segment/api.go

@@ -17,6 +17,8 @@ package segment
 import (
 	// standard libraries.
 	"context"
+	stderr "errors"
+	"io"
 
 	// third-party libraries.
 
@@ -137,46 +139,59 @@ func (s *segmentServer) AppendToBlock(
 }
 
 func (s *segmentServer) AppendToBlockStream(stream segpb.SegmentServer_AppendToBlockStreamServer) error {
-	ctx := context.Background()
+	ctx := stream.Context()
+	exitc := make(chan struct{})
 	for {
-		request, err := stream.Recv()
-		if err != nil {
-			log.Error(ctx, "append stream recv failed", map[string]interface{}{
-				log.KeyError: err,
-			})
-			return err
-		}
-
-		callbackFunc := func(offsets []int64, err error) {
-			errCode := errpb.ErrorCode_SUCCESS
-			errMsg := "success"
+		select {
+		case <-exitc:
+			return stderr.New("append stream send failed")
+		default:
+			request, err := stream.Recv()
 			if err != nil {
-				if errors.Is(err, errors.ErrFull) {
-					errCode = err.(*errors.ErrorType).Code
-					errMsg = err.(*errors.ErrorType).Message
-				} else {
-					errCode = errpb.ErrorCode_UNKNOWN
-					errMsg = "unknown"
+				if stderr.Is(err, io.EOF) {
+					log.Warning(ctx, "append stream closed", map[string]interface{}{
+						log.KeyError: err,
+					})
+					return nil
 				}
-				log.Error(ctx, "append to block failed", map[string]interface{}{
+				log.Error(ctx, "append stream recv failed", map[string]interface{}{
 					log.KeyError: err,
 				})
+				return err
 			}
 
-			err = stream.Send(&segpb.AppendToBlockStreamResponse{
-				Id:           request.Id,
-				ResponseCode: errCode,
-				ResponseMsg:  errMsg,
-				Offsets:      offsets,
-			})
-			if err != nil {
-				log.Error(ctx, "read stream send failed", map[string]interface{}{
-					log.KeyError: err,
+			callbackFunc := func(offsets []int64, err error) {
+				errCode := errpb.ErrorCode_SUCCESS
+				errMsg := "success"
+				if err != nil {
+					if errors.Is(err, errors.ErrFull) {
+						errCode = err.(*errors.ErrorType).Code
+						errMsg = err.(*errors.ErrorType).Message
+					} else {
+						errCode = errpb.ErrorCode_APPEND_TO_BLOCK_FAILED
+						errMsg = "append to block failed"
+					}
+					log.Error(ctx, "append to block failed", map[string]interface{}{
+						log.KeyError: err,
+					})
+				}
+
+				err = stream.Send(&segpb.AppendToBlockStreamResponse{
+					Id:           request.Id,
+					ResponseCode: errCode,
+					ResponseMsg:  errMsg,
+					Offsets:      offsets,
 				})
-				return
+				if err != nil {
+					log.Error(ctx, "append stream send failed", map[string]interface{}{
+						log.KeyError: err,
+					})
+					close(exitc)
+					return
+				}
 			}
+			s.srv.AppendToBlock(ctx, vanus.ID(request.BlockId), request.Events.Events, callbackFunc)
 		}
-		s.srv.AppendToBlock(ctx, vanus.ID(request.BlockId), request.Events.Events, callbackFunc)
 	}
 }
 
@@ -195,10 +210,16 @@ func (s *segmentServer) ReadFromBlock(
 }
 
 func (s *segmentServer) ReadFromBlockStream(stream segpb.SegmentServer_ReadFromBlockStreamServer) error {
-	ctx := context.Background()
+	ctx := stream.Context()
 	for {
 		request, err := stream.Recv()
 		if err != nil {
+			if stderr.Is(err, io.EOF) {
+				log.Warning(ctx, "read stream closed", map[string]interface{}{
+					log.KeyError: err,
+				})
+				return nil
+			}
 			log.Error(ctx, "read stream recv failed", map[string]interface{}{
 				log.KeyError: err,
 			})
@@ -211,8 +232,8 @@ func (s *segmentServer) ReadFromBlockStream(stream segpb.SegmentServer_ReadFromB
 		events, err := s.srv.ReadFromBlock(
 			ctx, blockID, request.Offset, int(request.Number), request.PollingTimeoutInMillisecond)
 		if err != nil {
-			errCode = errpb.ErrorCode_UNKNOWN
-			errMsg = "unknown"
+			errCode = errpb.ErrorCode_READ_FROM_BLOCK_FAILED
+			errMsg = "read from block failed"
 			log.Error(ctx, "read from block failed", map[string]interface{}{
 				log.KeyError: err,
 			})

internal/store/segment/server.go

@@ -72,7 +72,8 @@ const (
 	debugModeENV                = "SEGMENT_SERVER_DEBUG_MODE"
 	defaultLeaderInfoBufferSize = 256
 	defaultForceStopTimeout     = 30 * time.Second
-	defaultWaiterWorker         = 8
+	defaultCallbackWorker       = 8
+	defaultCallbackBufferSize   = 32
 )
 
 type Server interface {
@@ -111,7 +112,7 @@ func NewServer(cfg store.Config) Server {
 	srv := &server{
 		state:        primitive.ServerStateCreated,
 		cfg:          cfg,
-		callbackC:    make(chan func()),
+		callbackC:    make(chan func(), defaultCallbackBufferSize),
 		isDebugMode:  debugModel,
 		localAddress: localAddress,
 		volumeID:     uint64(cfg.Volume.ID),
@@ -218,7 +219,8 @@ func (s *server) preGrpcStream(ctx context.Context, info *tap.Info) (context.Con
 }
 
 func (s *server) runCallbackWorkPool(ctx context.Context) {
-	for i := 0; i < defaultWaiterWorker; i++ {
+	// TODO(jiangkai): use dynamic work pool
+	for i := 0; i < defaultCallbackWorker; i++ {
 		go s.runCallbackWorker(ctx)
 	}
 }
@@ -680,7 +682,9 @@ func (s *server) AppendToBlock(ctx context.Context, id vanus.ID, events []*cepb.
 	if v, ok := s.replicas.Load(id); ok {
 		b, _ = v.(Replica)
 	} else {
-		cb(nil, errors.ErrResourceNotFound.WithMessage("the block doesn't exist"))
+		s.callbackC <- func() {
+			cb(nil, errors.ErrResourceNotFound.WithMessage("the block doesn't exist"))
+		}
 		return
 	}
 
@@ -695,13 +699,18 @@ func (s *server) AppendToBlock(ctx context.Context, id vanus.ID, events []*cepb.
 	metrics.WriteThroughputCounterVec.WithLabelValues(s.volumeIDStr, b.IDStr()).Add(float64(size))
 
 	b.Append(ctx, entries, func(seqs []int64, err error) {
-		s.callbackC <- func() {
+		f := func() {
 			if err == nil {
 				// TODO(weihe.yin) make this method deep to code
 				s.pm.NewMessageArrived(id)
 			}
 			cb(seqs, s.processAppendError(ctx, b, err))
 		}
+		_, ok := <-s.closeC
+		if !ok {
+			f()
+		}
+		s.callbackC <- f
 	})
 }
 

pkg/errors/errors.go

@@ -24,12 +24,14 @@ var (
 
 	// INVALID_REQUEST
 	ErrInvalidRequest      = New("invalid request").WithCode(errpb.ErrorCode_INVALID_REQUEST)
-	ErrTransformInputParse = New("transform input invalid").WithCode(errpb.ErrorCode_TRANSFORM_INPUT_PARSE)
+	ErrTransformInputParse = New("transform input invalid").WithCode(errpb.ErrorCode_PARSE_TRANSFORM_INPUT)
 	ErrNoEndpoint          = New("no endpoint").WithCode(errpb.ErrorCode_NO_ENDPOINT)
 
 	// INTERNAL
 	ErrInternal       = New("internal error").WithCode(errpb.ErrorCode_INTERNAL)
 	ErrCorruptedEvent = New("corrupted event").WithCode(errpb.ErrorCode_CORRUPTED_EVENT)
+	ErrAppendToBlock  = New("append to block failed").WithCode(errpb.ErrorCode_APPEND_TO_BLOCK_FAILED)
+	ErrReadFromBlock  = New("read from block failed").WithCode(errpb.ErrorCode_READ_FROM_BLOCK_FAILED)
 
 	// FULL
 	ErrFull            = New("full").WithCode(errpb.ErrorCode_FULL)