revert tryagain error

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

Author: hwjiangkai

client/internal/vanus/store/block_store.go

@@ -23,22 +23,23 @@ import (
 	"io"
 	"sync"
 
-	"github.com/linkall-labs/vanus/observability/tracing"
 	"go.opentelemetry.io/otel/trace"
 
 	// third-party libraries.
 	ce "github.com/cloudevents/sdk-go/v2"
 	"github.com/scylladb/go-set/u64set"
 
-	// this project.
-	"github.com/linkall-labs/vanus/client/pkg/api"
-	"github.com/linkall-labs/vanus/client/pkg/eventlog"
-	"github.com/linkall-labs/vanus/client/pkg/policy"
+	// first-party libraries
 	vlog "github.com/linkall-labs/vanus/observability/log"
+	"github.com/linkall-labs/vanus/observability/tracing"
 	"github.com/linkall-labs/vanus/pkg/errors"
 
+	// this project.
 	eb "github.com/linkall-labs/vanus/client/internal/vanus/eventbus"
 	el "github.com/linkall-labs/vanus/client/internal/vanus/eventlog"
+	"github.com/linkall-labs/vanus/client/pkg/api"
+	"github.com/linkall-labs/vanus/client/pkg/eventlog"
+	"github.com/linkall-labs/vanus/client/pkg/policy"
 )
 
 func NewEventbus(cfg *eb.Config) *eventbus {

client/pkg/eventbus/eventbus.go

@@ -482,7 +482,9 @@ func (r *logReader) Read(ctx context.Context, size int16) ([]*ce.Event, error) {
 	if err != nil {
 		if errors.Is(err, errors.ErrOffsetOverflow) {
 			r.elog.refreshReadableSegments(ctx)
-			r.switchSegment(ctx)
+			if r.switchSegment(ctx) {
+				return nil, errors.ErrTryAgain
+			}
 		}
 		return nil, err
 	}
@@ -505,7 +507,9 @@ func (r *logReader) ReadStream(ctx context.Context, size int16) ([]*ce.Event, er
 	if err != nil {
 		if errors.Is(err, errors.ErrOffsetOverflow) {
 			r.elog.refreshReadableSegments(ctx)
-			r.switchSegment(ctx)
+			if r.switchSegment(ctx) {
+				return nil, errors.ErrTryAgain
+			}
 		}
 		return nil, err
 	}
@@ -522,10 +526,7 @@ func (r *logReader) selectReadableSegment(ctx context.Context) (*segment, error)
 	segment := func() *segment {
 		r.mu.RLock()
 		defer r.mu.RUnlock()
-		if r.cur != nil {
-			return r.cur
-		}
-		return nil
+		return r.cur
 	}()
 
 	if segment == nil {

client/pkg/eventlog/eventlog_impl.go

@@ -162,7 +162,7 @@ func (s *segment) Append(ctx context.Context, event *ce.Event) (int64, error) {
 
 	b := s.preferSegmentBlock()
 	if b == nil {
-		return -1, errors.ErrNotLeader.WithMessage("the block is not leader")
+		return -1, errors.ErrNoLeader.WithMessage("block no leader")
 	}
 	off, err := b.Append(_ctx, event)
 	if err != nil {
@@ -177,7 +177,7 @@ func (s *segment) AppendManyStream(ctx context.Context, events []*ce.Event) ([]i
 
 	b := s.preferSegmentBlock()
 	if b == nil {
-		return nil, errors.ErrNotLeader.WithMessage("the block is not leader")
+		return nil, errors.ErrNoLeader.WithMessage("block no leader")
 	}
 	offsets, err := b.AppendManyStream(_ctx, events)
 	if err != nil {

client/pkg/eventlog/log_segment.go

@@ -67,17 +67,13 @@ type peer struct {
 
 type LeaderChangedListener func(block, leader vanus.ID, term uint64)
 
-type CommitWaiter struct {
-	seqs     []int64
+type commitWaiter struct {
 	offset   int64
+	seqs     []int64
 	err      error
 	callback func([]int64, error)
 }
 
-func (w CommitWaiter) Do() {
-	w.callback(w.seqs, w.err)
-}
-
 type Appender interface {
 	block.Appender
 
@@ -92,8 +88,8 @@ type appender struct {
 	actx     block.AppendContext
 	appendMu sync.RWMutex
 
-	waiters      []CommitWaiter
-	waiterC      chan CommitWaiter
+	waiters      []commitWaiter
+	callbackC    chan func()
 	commitIndex  uint64
 	commitOffset int64
 	waitMu       sync.Mutex
@@ -122,21 +118,21 @@ func NewAppender(
 	raftLog *raftlog.Log,
 	host transport.Host,
 	listener LeaderChangedListener,
-	waiterC chan CommitWaiter,
+	callbackC chan func(),
 ) Appender {
 	ctx, cancel := context.WithCancel(ctx)
 
 	a := &appender{
-		raw:      raw,
-		waiters:  make([]CommitWaiter, 0),
-		waiterC:  waiterC,
-		listener: listener,
-		log:      raftLog,
-		host:     host,
-		hint:     make([]peer, 0, defaultHintCapacity),
-		cancel:   cancel,
-		doneC:    make(chan struct{}),
-		tracer:   tracing.NewTracer("store.block.raft.appender", trace.SpanKindInternal),
+		raw:       raw,
+		waiters:   make([]commitWaiter, 0),
+		callbackC: callbackC,
+		listener:  listener,
+		log:       raftLog,
+		host:      host,
+		hint:      make([]peer, 0, defaultHintCapacity),
+		cancel:    cancel,
+		doneC:     make(chan struct{}),
+		tracer:    tracing.NewTracer("store.block.raft.appender", trace.SpanKindInternal),
 	}
 	a.actx = a.raw.NewAppendContext(nil)
 	a.commitOffset = a.actx.WriteOffset()
@@ -489,9 +485,9 @@ func (a *appender) Append(ctx context.Context, entries []block.Entry, cb func([]
 	}
 
 	// register callback and wait until entries is committed.
-	a.registerCommitWaiter(ctx, CommitWaiter{
-		seqs:     seqs,
+	a.registerCommitWaiter(ctx, commitWaiter{
 		offset:   offset,
+		seqs:     seqs,
 		err:      err,
 		callback: cb,
 	})
@@ -538,20 +534,23 @@ func (a *appender) doAppend(ctx context.Context, frags ...block.Fragment) {
 	_, _ = a.raw.CommitAppend(ctx, frags...)
 }
 
-func (a *appender) registerCommitWaiter(ctx context.Context, waiter CommitWaiter) {
+func (a *appender) registerCommitWaiter(ctx context.Context, waiter commitWaiter) {
 	_, span := a.tracer.Start(ctx, "waitCommit")
 	defer span.End()
 
 	span.AddEvent("Acquiring wait lock")
 	a.waitMu.Lock()
-	defer a.waitMu.Unlock()
 	span.AddEvent("Got wait lock")
 
 	if waiter.offset <= a.commitOffset {
-		waiter.callback(waiter.seqs, waiter.err)
+		a.waitMu.Unlock()
+		a.callbackC <- func() {
+			waiter.callback(waiter.seqs, waiter.err)
+		}
 		return
 	}
 	a.waiters = append(a.waiters, waiter)
+	a.waitMu.Unlock()
 }
 
 func (a *appender) doWakeup(ctx context.Context, commit int64) {
@@ -569,7 +568,9 @@ func (a *appender) doWakeup(ctx context.Context, commit int64) {
 		if waiter.offset > commit {
 			break
 		}
-		a.waiterC <- waiter
+		a.callbackC <- func() {
+			waiter.callback(waiter.seqs, waiter.err)
+		}
 		a.waiters = a.waiters[1:]
 	}
 	a.commitOffset = commit

internal/store/block/raft/appender.go

@@ -130,7 +130,7 @@ func (s *segmentServer) AppendToBlock(
 	s.srv.AppendToBlock(ctx, blockID, events, func(offs []int64, e error) {
 		offsets = offs
 		err = e
-		donec <- struct{}{}
+		close(donec)
 	})
 	<-donec
 	return &segpb.AppendToBlockResponse{Offsets: offsets}, err
@@ -163,10 +163,8 @@ func (s *segmentServer) AppendToBlockStream(stream segpb.SegmentServer_AppendToB
 				})
 			}
 
-			s.srv.NewMessageArrived(ctx, vanus.ID(request.BlockId))
-
 			err = stream.Send(&segpb.AppendToBlockStreamResponse{
-				ResponseId:   request.RequestId,
+				Id:           request.Id,
 				ResponseCode: errCode,
 				ResponseMsg:  errMsg,
 				Offsets:      offsets,
@@ -221,7 +219,7 @@ func (s *segmentServer) ReadFromBlockStream(stream segpb.SegmentServer_ReadFromB
 		}
 
 		err = stream.Send(&segpb.ReadFromBlockStreamResponse{
-			ResponseId:   request.RequestId,
+			Id:           request.Id,
 			ResponseCode: errCode,
 			ResponseMsg:  errMsg,
 			Events:       &cepb.CloudEventBatch{Events: events},

internal/store/segment/api.go

@@ -78,7 +78,7 @@ func (s *server) recoverBlocks(ctx context.Context, logs map[vanus.ID]*raftlog.L
 				return err
 			}
 		}
-		a := raft.NewAppender(context.TODO(), r, l, s.host, s.leaderChanged, s.waiterC)
+		a := raft.NewAppender(context.TODO(), r, l, s.host, s.leaderChanged, s.callbackC)
 		s.replicas.Store(id, &replica{
 			id:       id,
 			idStr:    id.String(),

internal/store/segment/recovery.go

@@ -116,7 +116,7 @@ func (s *server) createBlock(ctx context.Context, id vanus.ID, size int64) (Repl
 
 	// Create replica.
 	l := raftlog.NewLog(id, s.wal, s.metaStore, s.offsetStore, nil)
-	a := raft.NewAppender(context.TODO(), r, l, s.host, s.leaderChanged, s.waiterC)
+	a := raft.NewAppender(context.TODO(), r, l, s.host, s.leaderChanged, s.callbackC)
 
 	return &replica{
 		id:       id,

internal/store/segment/replica.go

@@ -93,8 +93,6 @@ type Server interface {
 	AppendToBlock(ctx context.Context, id vanus.ID, events []*cepb.CloudEvent, cb func([]int64, error))
 	ReadFromBlock(ctx context.Context, id vanus.ID, seq int64, num int, pollingTimeout uint32) ([]*cepb.CloudEvent, error)
 	LookupOffsetInBlock(ctx context.Context, id vanus.ID, stime int64) (int64, error)
-
-	NewMessageArrived(ctx context.Context, id vanus.ID)
 }
 
 func NewServer(cfg store.Config) Server {
@@ -112,7 +110,7 @@ func NewServer(cfg store.Config) Server {
 	srv := &server{
 		state:        primitive.ServerStateCreated,
 		cfg:          cfg,
-		waiterC:      make(chan raft.CommitWaiter),
+		callbackC:    make(chan func()),
 		isDebugMode:  debugModel,
 		localAddress: localAddress,
 		volumeID:     uint64(cfg.Volume.ID),
@@ -163,7 +161,7 @@ type server struct {
 	ctrl        cluster.Cluster
 	cc          ctrlpb.SegmentControllerClient
 	leaderC     chan leaderInfo
-	waiterC     chan raft.CommitWaiter
+	callbackC   chan func()
 
 	grpcSrv *grpc.Server
 	closeC  chan struct{}
@@ -218,19 +216,20 @@ func (s *server) preGrpcStream(ctx context.Context, info *tap.Info) (context.Con
 	return ctx, nil
 }
 
-func (s *server) runWaiterWorker(ctx context.Context) {
+func (s *server) runCallbackWorkPool(ctx context.Context) {
 	for i := 0; i < defaultWaiterWorker; i++ {
-		go func() {
-			for {
-				select {
-				case waiter := <-s.waiterC:
-					waiter.Do()
-				case <-ctx.Done():
-					close(s.waiterC)
-					return
-				}
-			}
-		}()
+		go s.runCallbackWorker(ctx)
+	}
+}
+
+func (s *server) runCallbackWorker(_ context.Context) {
+	for {
+		select {
+		case cb := <-s.callbackC:
+			cb()
+		case <-s.closeC:
+			return
+		}
 	}
 }
 
@@ -265,7 +264,7 @@ func (s *server) Initialize(ctx context.Context) error {
 		s.state = primitive.ServerStateRunning
 	}
 
-	s.runWaiterWorker(ctx)
+	s.runCallbackWorkPool(ctx)
 
 	return nil
 }
@@ -688,14 +687,12 @@ func (s *server) AppendToBlock(ctx context.Context, id vanus.ID, events []*cepb.
 	metrics.WriteTPSCounterVec.WithLabelValues(s.volumeIDStr, b.IDStr()).Add(float64(len(events)))
 	metrics.WriteThroughputCounterVec.WithLabelValues(s.volumeIDStr, b.IDStr()).Add(float64(size))
 
-	b.Append(ctx, entries, cb)
-
-	// TODO(weihe.yin) make this method deep to code
-	// s.pm.NewMessageArrived(id)
-}
-
-func (s *server) NewMessageArrived(_ context.Context, id vanus.ID) {
-	s.pm.NewMessageArrived(id)
+	b.Append(ctx, entries, func(offsets []int64, err error) {
+		if err == nil {
+			s.pm.NewMessageArrived(id)
+		}
+		cb(offsets, err)
+	})
 }
 
 func (s *server) onBlockArchived(stat block.Statistics) {

internal/store/segment/server.go

@@ -199,7 +199,7 @@ func (b *bucket) run(ctx context.Context) {
 				// batch read
 				events, err := b.getEvent(ctx, defaultNumberOfEventsRead)
 				if err != nil {
-					if !errors.Is(err, errors.ErrOffsetOnEnd) {
+					if !errors.Is(err, errors.ErrOffsetOnEnd) && !errors.Is(err, errors.ErrTryAgain) {
 						log.Error(ctx, "get event failed when bucket running", map[string]interface{}{
 							"eventbus":   b.getEventbus(),
 							log.KeyError: err,

internal/timer/timingwheel/bucket.go

@@ -286,7 +286,7 @@ func (elReader *eventLogReader) run(ctx context.Context) {
 		for {
 			err = elReader.readEvent(ctx, lr)
 			switch {
-			case err == nil, errors.Is(err, errors.ErrOffsetOnEnd), errors.Is(err, errors.ErrOffsetOverflow):
+			case err == nil, errors.Is(err, errors.ErrOffsetOnEnd), errors.Is(err, errors.ErrTryAgain):
 			case stderr.Is(err, context.Canceled):
 				return
 			case errors.Is(err, errors.ErrOffsetUnderflow):

internal/trigger/reader/reader.go

@@ -41,7 +41,7 @@ func TestCtrlClientIsNeedRetry(t *testing.T) {
 			err = errors.New("xxxxx: 1111111111 ")
 			So(isNeedRetry(err), ShouldBeFalse)
 
-			err = errors.New("balabala, please connect to: 127.0.0.1:2048 ").WithGRPCCode(errpb.ErrorCode_NOT_LEADER)
+			err = errors.New("balabala, please connect to: 127.0.0.1:2048 ").WithCode(errpb.ErrorCode_NOT_LEADER)
 			So(isNeedRetry(err), ShouldBeTrue)
 		})
 	})

pkg/cluster/raw_client/segment_test.go

@@ -42,7 +42,7 @@ type ErrorType struct {
 	underlayErrors []error
 }
 
-func (e *ErrorType) WithGRPCCode(c errpb.ErrorCode) *ErrorType {
+func (e *ErrorType) WithCode(c errpb.ErrorCode) *ErrorType {
 	_e := e.copy()
 	_e.Code = c
 	return _e