feat: optimize route switching when freeze segment

client/internal/vanus/eventlog/name_service.go

@@ -48,13 +48,13 @@ type NameService struct {
 	tracer *tracing.Tracer
 }
 
-func (ns *NameService) LookupWritableSegment(ctx context.Context, logID uint64) (*record.Segment, error) {
+func (ns *NameService) LookupWritableSegment(ctx context.Context, logID uint64) ([]*record.Segment, error) {
 	ctx, span := ns.tracer.Start(ctx, "LookupWritableSegment")
 	defer span.End()
 
 	req := &ctrlpb.GetAppendableSegmentRequest{
 		EventLogId: logID,
-		Limited:    1,
+		Limited:    2,
 	}
 
 	resp, err := ns.client.GetAppendableSegment(ctx, req)
@@ -75,26 +75,28 @@ func (ns *NameService) LookupWritableSegment(ctx context.Context, logID uint64)
 	if len(segments) == 0 {
 		return nil, errors.ErrNotWritable
 	}
-	return segments[0], nil
+	return segments, nil
 }
 
 func (ns *NameService) LookupReadableSegments(ctx context.Context, logID uint64) ([]*record.Segment, error) {
 	ctx, span := ns.tracer.Start(ctx, "LookupReadableSegments")
 	defer span.End()
 
-	req := &ctrlpb.ListSegmentRequest{
-		EventLogId:  logID,
-		StartOffset: 0,
-		EndOffset:   math.MaxInt64,
-		Limited:     math.MaxInt32,
+	req := &ctrlpb.GetReadableSegmentRequest{
+		EventLogId: logID,
+		Limited:    math.MaxInt32,
 	}
 
-	resp, err := ns.client.ListSegment(ctx, req)
+	resp, err := ns.client.GetReadableSegment(ctx, req)
 	if err != nil {
 		return nil, err
 	}
 
 	segments := toSegments(resp.GetSegments())
+	if len(segments) == 0 {
+		return nil, errors.ErrNotReadable
+	}
+
 	return segments, nil
 }
 
@@ -107,9 +109,9 @@ func toSegments(pbs []*metapb.Segment) []*record.Segment {
 		segment := toSegment(pb)
 		segments = append(segments, segment)
 		// only return first working segment
-		if segment.Writable {
-			break
-		}
+		// if segment.Writable {
+		// 	break
+		// }
 	}
 	return segments
 }
@@ -123,13 +125,15 @@ func toSegment(segment *metapb.Segment) *record.Segment {
 		}
 	}
 	return &record.Segment{
-		ID:               segment.GetId(),
-		StartOffset:      segment.GetStartOffsetInLog(),
-		EndOffset:        segment.GetEndOffsetInLog(),
-		FirstEventBornAt: time.UnixMilli(segment.FirstEventBornAtByUnixMs),
-		LastEventBornAt:  time.UnixMilli(segment.LastEvnetBornAtByUnixMs),
-		Writable:         segment.State == "working", // TODO: writable
-		Blocks:           blocks,
-		LeaderBlockID:    segment.GetLeaderBlockId(),
+		ID:                segment.GetId(),
+		PreviousSegmentId: segment.GetPreviousSegmentId(),
+		NextSegmentId:     segment.GetNextSegmentId(),
+		StartOffset:       segment.GetStartOffsetInLog(),
+		EndOffset:         segment.GetEndOffsetInLog(),
+		FirstEventBornAt:  time.UnixMilli(segment.FirstEventBornAtByUnixMs),
+		LastEventBornAt:   time.UnixMilli(segment.LastEvnetBornAtByUnixMs),
+		Writable:          segment.State == "working", // TODO: writable
+		Blocks:            blocks,
+		LeaderBlockID:     segment.GetLeaderBlockId(),
 	}
 }

client/pkg/api/client.go

@@ -17,6 +17,7 @@ package api
 
 import (
 	"context"
+
 	ce "github.com/cloudevents/sdk-go/v2"
 	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
 )
@@ -32,7 +33,7 @@ type Eventbus interface {
 
 type BusWriter interface {
 	AppendOne(ctx context.Context, event *ce.Event, opts ...WriteOption) (eid string, err error)
-	AppendMany(ctx context.Context, events []*ce.Event, opts ...WriteOption) (eid string, err error)
+	AppendMany(ctx context.Context, events []*ce.Event, opts ...WriteOption) (err error)
 	AppendBatch(ctx context.Context, events *cloudevents.CloudEventBatch, opts ...WriteOption) (err error)
 }
 

client/pkg/eventbus/eventbus.go

@@ -17,13 +17,12 @@ package eventbus
 import (
 	// standard libraries.
 	"context"
-	"encoding/base64"
-	"encoding/binary"
 	stderrors "errors"
-	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
 	"io"
 	"sync"
 
+	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
+
 	"github.com/linkall-labs/vanus/observability/tracing"
 	"go.opentelemetry.io/otel/trace"
 
@@ -37,7 +36,7 @@ import (
 	"github.com/linkall-labs/vanus/client/pkg/api"
 	"github.com/linkall-labs/vanus/client/pkg/eventlog"
 	"github.com/linkall-labs/vanus/client/pkg/policy"
-	vlog "github.com/linkall-labs/vanus/observability/log"
+	"github.com/linkall-labs/vanus/observability/log"
 	"github.com/linkall-labs/vanus/pkg/errors"
 
 	eb "github.com/linkall-labs/vanus/client/internal/vanus/eventbus"
@@ -67,7 +66,7 @@ func NewEventbus(cfg *eb.Config) *eventbus {
 		for {
 			re, ok := <-ch
 			if !ok {
-				vlog.Debug(context.Background(), "eventbus quits writable watcher", map[string]interface{}{
+				log.Debug(context.Background(), "eventbus quits writable watcher", map[string]interface{}{
 					"eventbus": bus.cfg.Name,
 				})
 				break
@@ -89,7 +88,7 @@ func NewEventbus(cfg *eb.Config) *eventbus {
 		for {
 			re, ok := <-ch
 			if !ok {
-				vlog.Debug(context.Background(), "eventbus quits readable watcher", map[string]interface{}{
+				log.Debug(context.Background(), "eventbus quits readable watcher", map[string]interface{}{
 					"eventbus": bus.cfg.Name,
 				})
 				break
@@ -185,19 +184,23 @@ func (b *eventbus) GetLog(ctx context.Context, logID uint64, opts ...api.LogOpti
 	}
 
 	if op.Policy.AccessMode() == api.ReadOnly {
+		b.readableMu.RLock()
+		defer b.readableMu.RUnlock()
 		if len(b.readableLogs) == 0 {
 			b.refreshReadableLogs(ctx)
 		}
-		if log, ok := b.readableLogs[logID]; ok {
-			return log, nil
+		if l, ok := b.readableLogs[logID]; ok {
+			return l, nil
 		}
 		return nil, errors.ErrResourceNotFound.WithMessage("eventlog not found")
 	} else if op.Policy.AccessMode() == api.ReadWrite {
+		b.writableMu.RLock()
+		defer b.writableMu.RUnlock()
 		if len(b.writableLogs) == 0 {
 			b.refreshWritableLogs(ctx)
 		}
-		if log, ok := b.writableLogs[logID]; ok {
-			return log, nil
+		if l, ok := b.writableLogs[logID]; ok {
+			return l, nil
 		}
 		return nil, errors.ErrResourceNotFound.WithMessage("eventlog not found")
 	} else {
@@ -216,6 +219,8 @@ func (b *eventbus) ListLog(ctx context.Context, opts ...api.LogOption) ([]api.Ev
 	}
 
 	if op.Policy.AccessMode() == api.ReadOnly {
+		b.readableMu.RLock()
+		defer b.readableMu.RUnlock()
 		if len(b.readableLogs) == 0 {
 			b.refreshReadableLogs(ctx)
 		}
@@ -225,6 +230,8 @@ func (b *eventbus) ListLog(ctx context.Context, opts ...api.LogOption) ([]api.Ev
 		}
 		return eventlogs, nil
 	} else if op.Policy.AccessMode() == api.ReadWrite {
+		b.writableMu.RLock()
+		defer b.writableMu.RUnlock()
 		if len(b.writableLogs) == 0 {
 			b.refreshWritableLogs(ctx)
 		}
@@ -311,8 +318,8 @@ func (b *eventbus) updateWritableLogs(ctx context.Context, re *WritableLogsResul
 			Endpoints: b.cfg.Endpoints,
 			ID:        logID,
 		}
-		log := eventlog.NewEventLog(cfg)
-		lws[logID] = log
+		l := eventlog.NewEventLog(cfg)
+		lws[logID] = l
 		return true
 	})
 	b.setWritableLogs(s, lws)
@@ -404,8 +411,8 @@ func (b *eventbus) updateReadableLogs(ctx context.Context, re *ReadableLogsResul
 			Endpoints: b.cfg.Endpoints,
 			ID:        logID,
 		}
-		log := eventlog.NewEventLog(cfg)
-		lws[logID] = log
+		l := eventlog.NewEventLog(cfg)
+		lws[logID] = l
 		return true
 	})
 	b.setReadableLogs(s, lws)
@@ -471,7 +478,7 @@ func (w *busWriter) AppendBatch(ctx context.Context, events *cloudevents.CloudEv
 
 var _ api.BusWriter = (*busWriter)(nil)
 
-func (w *busWriter) AppendOne(ctx context.Context, event *ce.Event, opts ...api.WriteOption) (eid string, err error) {
+func (w *busWriter) AppendOne(ctx context.Context, event *ce.Event, opts ...api.WriteOption) (string, error) {
 	_ctx, span := w.tracer.Start(ctx, "AppendOne")
 	defer span.End()
 
@@ -490,23 +497,17 @@ func (w *busWriter) AppendOne(ctx context.Context, event *ce.Event, opts ...api.
 	}
 
 	// 2. append the event to the eventlog
-	off, err := lw.Append(_ctx, event)
+	eid, err := lw.Append(_ctx, event)
 	if err != nil {
 		return "", err
 	}
 
-	// 3. generate event ID
-	var buf [16]byte
-	binary.BigEndian.PutUint64(buf[0:8], lw.Log().ID())
-	binary.BigEndian.PutUint64(buf[8:16], uint64(off))
-	encoded := base64.StdEncoding.EncodeToString(buf[:])
-
-	return encoded, nil
+	return eid, nil
 }
 
-func (w *busWriter) AppendMany(ctx context.Context, events []*ce.Event, opts ...api.WriteOption) (eid string, err error) {
+func (w *busWriter) AppendMany(ctx context.Context, events []*ce.Event, opts ...api.WriteOption) (err error) {
 	// TODO(jiangkai): implement this method, by jiangkai, 2022.10.24
-	return "", nil
+	return nil
 }
 
 func (w *busWriter) Bus() api.Eventbus {
@@ -517,17 +518,17 @@ func (w *busWriter) pickWritableLog(ctx context.Context, opts *api.WriteOptions)
 	_ctx, span := w.tracer.Start(ctx, "pickWritableLog")
 	defer span.End()
 
-	log, err := opts.Policy.NextLog(ctx)
+	l, err := opts.Policy.NextLog(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	l := w.ebus.getWritableLog(_ctx, log.ID())
-	if l == nil {
+	lw := w.ebus.getWritableLog(_ctx, l.ID())
+	if lw == nil {
 		return nil, stderrors.New("can not pick writable log")
 	}
 
-	return l.Writer(), nil
+	return lw.Writer(), nil
 }
 
 type busReader struct {
@@ -578,11 +579,11 @@ func (r *busReader) pickReadableLog(ctx context.Context, opts *api.ReadOptions)
 	_ctx, span := r.tracer.Start(ctx, "pickReadableLog")
 	defer span.End()
 
-	log, err := opts.Policy.NextLog(ctx)
+	l, err := opts.Policy.NextLog(ctx)
 	if err != nil {
 		return nil, err
 	}
-	lr := r.ebus.getReadableLog(_ctx, log.ID())
+	lr := r.ebus.getReadableLog(_ctx, l.ID())
 	if lr == nil {
 		return nil, stderrors.New("can not pick readable log")
 	}

client/pkg/eventlog/eventlog.go

@@ -18,6 +18,7 @@ package eventlog
 import (
 	// standard libraries.
 	"context"
+
 	"github.com/linkall-labs/vanus/proto/pkg/cloudevents"
 
 	// third-party libraries.
@@ -50,7 +51,7 @@ type LogWriter interface {
 
 	Close(ctx context.Context)
 
-	Append(ctx context.Context, event *ce.Event) (off int64, err error)
+	Append(ctx context.Context, event *ce.Event) (eid string, err error)
 	AppendMany(ctx context.Context, events *cloudevents.CloudEventBatch) (off int64, err error)
 }