Merge pull request #1040 Fixed topic writer hand up on server temporary error

Author: rekby

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed handle of operational errors in topic streams
 * The minimum version of Go in `ydb-go-sdk` has been raised to `go1.21`
 * Fixed topic writer infinite reconnections in some cases
 * Refactored nil on err `internal/grpcwrapper/rawydb/issues.go`, when golangci-lint nilerr enabled

go.mod

@@ -15,7 +15,7 @@ require (
 
 // requires for tests only
 require (
-	github.com/rekby/fixenv v0.3.2
+	github.com/rekby/fixenv v0.6.1
 	github.com/stretchr/testify v1.7.1
 	go.uber.org/mock v0.3.1-0.20231011042131-892b665398ec // indirect
 )

go.sum

@@ -60,6 +60,8 @@ github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZN
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/rekby/fixenv v0.3.2 h1:6AOdQ9Boaa/lOQJTY8GDmQRIhg3S3SD0mIEPkuDSkoQ=
 github.com/rekby/fixenv v0.3.2/go.mod h1:/b5LRc06BYJtslRtHKxsPWFT/ySpHV+rWvzTg+XWk4c=
+github.com/rekby/fixenv v0.6.1 h1:jUFiSPpajT4WY2cYuc++7Y1zWrnCxnovGCIX72PZniM=
+github.com/rekby/fixenv v0.6.1/go.mod h1:/b5LRc06BYJtslRtHKxsPWFT/ySpHV+rWvzTg+XWk4c=
 github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=

internal/grpcwrapper/rawtopic/rawtopicreader/rawtopicreader.go

@@ -31,7 +31,9 @@ func (s StreamReader) CloseSend() error {
 func (s StreamReader) Recv() (ServerMessage, error) {
 	grpcMess, err := s.Stream.Recv()
 	if err != nil {
-		err = xerrors.Transport(err)
+		if !xerrors.IsErrorFromServer(err) {
+			err = xerrors.Transport(err)
+		}
 
 		return nil, err
 	}

internal/grpcwrapper/rawtopic/rawtopicwriter/streamwriter.go

@@ -41,7 +41,9 @@ func (w *StreamWriter) Recv() (ServerMessage, error) {
 
 	grpcMsg, err := w.Stream.Recv()
 	if err != nil {
-		err = xerrors.Transport(err)
+		if !xerrors.IsErrorFromServer(err) {
+			err = xerrors.Transport(err)
+		}
 
 		return nil, xerrors.WithStackTrace(xerrors.Wrap(fmt.Errorf(
 			"ydb: failed to read grpc message from writer stream: %w",

internal/topic/topicwriterinternal/writer_grpc_mock_test.go

@@ -0,0 +1,134 @@
+package topicwriterinternal_test
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/rekby/fixenv"
+	"github.com/rekby/fixenv/sf"
+	"github.com/stretchr/testify/require"
+	"github.com/ydb-platform/ydb-go-genproto/Ydb_Topic_V1"
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Issue"
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Topic"
+
+	"github.com/ydb-platform/ydb-go-sdk/v3"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xtest"
+	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions"
+	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicwriter"
+)
+
+func TestRegressionOperationUnavailableIssue1007(t *testing.T) {
+	xtest.TestManyTimes(t, func(t testing.TB) {
+		e := fixenv.New(t)
+
+		mock := newTopicWriterOperationUnavailable()
+		connString := xtest.GrpcMockTopicConnString(e, mock)
+
+		db, err := ydb.Open(sf.Context(e), connString)
+		require.NoError(t, err)
+
+		writer, err := db.Topic().StartWriter("test", topicoptions.WithWriterWaitServerAck(true))
+		require.NoError(t, err)
+
+		err = writer.Write(sf.Context(e), topicwriter.Message{
+			Data: strings.NewReader("asd"),
+		})
+		require.NoError(t, err)
+		require.True(t, mock.UnavailableResponsed)
+	})
+}
+
+type topicWriterOperationUnavailable struct {
+	Ydb_Topic_V1.UnimplementedTopicServiceServer
+
+	UnavailableResponsed bool
+}
+
+func newTopicWriterOperationUnavailable() *topicWriterOperationUnavailable {
+	return &topicWriterOperationUnavailable{}
+}
+
+func (t *topicWriterOperationUnavailable) StreamWrite(server Ydb_Topic_V1.TopicService_StreamWriteServer) error {
+	initMsg, err := server.Recv()
+	if err != nil {
+		return fmt.Errorf("failed read init message: %w", err)
+	}
+
+	if initMsg.GetInitRequest() == nil {
+		return errors.New("first message must be init message")
+	}
+
+	err = server.Send(&Ydb_Topic.StreamWriteMessage_FromServer{
+		Status: Ydb.StatusIds_SUCCESS,
+		ServerMessage: &Ydb_Topic.StreamWriteMessage_FromServer_InitResponse{
+			InitResponse: &Ydb_Topic.StreamWriteMessage_InitResponse{
+				LastSeqNo:       0,
+				SessionId:       "test",
+				PartitionId:     0,
+				SupportedCodecs: nil,
+			},
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("failed to send init response: %w", err)
+	}
+
+	if !t.UnavailableResponsed {
+		t.UnavailableResponsed = true
+
+		err = server.Send(&Ydb_Topic.StreamWriteMessage_FromServer{
+			Status: Ydb.StatusIds_UNAVAILABLE,
+			Issues: []*Ydb_Issue.IssueMessage{
+				{
+					Message: "Test status unavailable",
+				},
+			},
+		})
+
+		if err != nil {
+			return fmt.Errorf("failed to send error response: %w", err)
+		}
+
+		return nil
+	}
+
+	// wait message block
+	messagesMsg, err := server.Recv()
+	if err != nil {
+		return errors.New("failed to read messages block")
+	}
+
+	if len(messagesMsg.GetClientMessage().(*Ydb_Topic.StreamWriteMessage_FromClient_WriteRequest).
+		WriteRequest.GetMessages()) == 0 {
+		return errors.New("received zero messages block")
+	}
+
+	err = server.Send(&Ydb_Topic.StreamWriteMessage_FromServer{
+		Status: Ydb.StatusIds_SUCCESS,
+		ServerMessage: &Ydb_Topic.StreamWriteMessage_FromServer_WriteResponse{
+			WriteResponse: &Ydb_Topic.StreamWriteMessage_WriteResponse{
+				Acks: []*Ydb_Topic.StreamWriteMessage_WriteResponse_WriteAck{
+					{
+						SeqNo: 1,
+						MessageWriteStatus: &Ydb_Topic.StreamWriteMessage_WriteResponse_WriteAck_Written_{
+							Written: &Ydb_Topic.StreamWriteMessage_WriteResponse_WriteAck_Written{
+								Offset: 1,
+							},
+						},
+					},
+				},
+				PartitionId:     0,
+				WriteStatistics: &Ydb_Topic.StreamWriteMessage_WriteResponse_WriteStatistics{},
+			},
+		},
+	})
+
+	if err != nil {
+		return fmt.Errorf("failed to sent write ack: %w", err)
+	}
+
+	return nil
+}

internal/xerrors/xerrors.go

@@ -79,6 +79,12 @@ func As(err error, targets ...interface{}) bool {
 	return false
 }
 
+// IsErrorFromServer return true if err returned from server
+// (opposite to raised internally in sdk)
+func IsErrorFromServer(err error) bool {
+	return IsTransportError(err) || IsOperationError(err)
+}
+
 // Is is a improved proxy to errors.Is
 // This need to single import errors
 func Is(err error, targets ...error) bool {

internal/xtest/ydb_grpc_mocks.go

@@ -0,0 +1,140 @@
+package xtest
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"reflect"
+	"strconv"
+	"time"
+
+	"github.com/rekby/fixenv"
+	"github.com/rekby/fixenv/sf"
+	"github.com/ydb-platform/ydb-go-genproto/Ydb_Discovery_V1"
+	"github.com/ydb-platform/ydb-go-genproto/Ydb_Topic_V1"
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Discovery"
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Operations"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/anypb"
+)
+
+func GrpcMockTopicConnString(e fixenv.Env, topicServiceImpl Ydb_Topic_V1.TopicServiceServer) string {
+	v := reflect.ValueOf(topicServiceImpl)
+	addr := v.Pointer()
+
+	var f fixenv.GenericFixtureFunction[string] = func() (*fixenv.GenericResult[string], error) {
+		listener := sf.LocalTCPListenerNamed(e, fmt.Sprintf("ydb-grpc-mock-topic-%v", addr))
+		connString := fmt.Sprintf("grpc://%s/local", listener.Addr())
+
+		mock, err := newGrpcMock(listener, topicServiceImpl)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create grpc mock: %w", err)
+		}
+
+		clean := func() {
+			_ = mock.Close()
+		}
+
+		return fixenv.NewGenericResultWithCleanup(connString, clean), nil
+	}
+
+	return fixenv.CacheResult(e, f, fixenv.CacheOptions{CacheKey: addr})
+}
+
+type grpcMock struct {
+	listener   net.Listener
+	grpcServer *grpc.Server
+	stopChan   chan error
+}
+
+func (m *grpcMock) Close() error {
+	m.grpcServer.Stop()
+
+	return m.listener.Close()
+}
+
+func newGrpcMock(listener net.Listener, topicServiceImpl Ydb_Topic_V1.TopicServiceServer) (*grpcMock, error) {
+	res := &grpcMock{
+		listener:   listener,
+		grpcServer: grpc.NewServer(),
+		stopChan:   make(chan error, 1),
+	}
+
+	host, portS, err := net.SplitHostPort(listener.Addr().String())
+	if err != nil {
+		return nil, fmt.Errorf("failed to split host port addresses: %w", err)
+	}
+
+	port, err := strconv.ParseUint(portS, 10, 32)
+	if err != nil {
+		return nil, fmt.Errorf("failed convert port to int: %w", err)
+	}
+	discoveryService := newMockDiscoveryService(host, uint32(port))
+
+	Ydb_Discovery_V1.RegisterDiscoveryServiceServer(res.grpcServer, discoveryService)
+	Ydb_Topic_V1.RegisterTopicServiceServer(res.grpcServer, topicServiceImpl)
+
+	go func() {
+		res.stopChan <- res.grpcServer.Serve(res.listener)
+	}()
+
+	select {
+	case <-res.stopChan:
+		return nil, err
+	case <-time.After(time.Millisecond):
+		return res, nil
+	}
+}
+
+type mockDiscoveryService struct {
+	Ydb_Discovery_V1.UnimplementedDiscoveryServiceServer
+	host string
+	port uint32
+}
+
+func newMockDiscoveryService(host string, port uint32) *mockDiscoveryService {
+	return &mockDiscoveryService{
+		host: host,
+		port: port,
+	}
+}
+
+func (m mockDiscoveryService) ListEndpoints(
+	ctx context.Context,
+	request *Ydb_Discovery.ListEndpointsRequest,
+) (*Ydb_Discovery.ListEndpointsResponse, error) {
+	res := &Ydb_Discovery.ListEndpointsResult{
+		Endpoints: []*Ydb_Discovery.EndpointInfo{
+			{
+				Address:    m.host,
+				Port:       m.port,
+				LoadFactor: 0,
+				Ssl:        false,
+				Service:    nil,
+				Location:   "",
+				NodeId:     1,
+				IpV4:       []string{"127.0.0.1"},
+			},
+		},
+		SelfLocation: "",
+	}
+	resp := &Ydb_Discovery.ListEndpointsResponse{
+		Operation: &Ydb_Operations.Operation{
+			Id:     "test-list-operation",
+			Ready:  true,
+			Status: Ydb.StatusIds_SUCCESS,
+			Result: &anypb.Any{},
+		},
+	}
+	err := resp.GetOperation().GetResult().MarshalFrom(res)
+
+	return resp, err
+}
+
+func (m mockDiscoveryService) WhoAmI(
+	ctx context.Context,
+	request *Ydb_Discovery.WhoAmIRequest,
+) (*Ydb_Discovery.WhoAmIResponse, error) {
+	panic("unimplemented")
+}

tests/integration/helpers_test.go

@@ -67,6 +67,16 @@ func (scope *scopeT) Failed() bool {
 	return scope.t.Failed()
 }
 
+// CacheWithCleanup wrap new interface as deprecated - for prevent re-write a lot of code
+// in new code prefer to use generic version of cache: fixenv.CacheResult(env, ...)
+func (scope *scopeT) CacheWithCleanup(cacheKey interface{}, opt *fixenv.FixtureOptions, f fixenv.FixtureCallbackWithCleanupFunc) interface{} {
+	fWrap := func() (*fixenv.Result, error) {
+		res, cleanup, err := f()
+		return fixenv.NewResultWithCleanup(res, cleanup), err
+	}
+	return scope.CacheResult(fWrap)
+}
+
 func (scope *scopeT) ConnectionString() string {
 	if envString := os.Getenv("YDB_CONNECTION_STRING"); envString != "" {
 		return envString