small fixes

Author: rekby

internal/grpcwrapper/rawtopic/client.go

@@ -3,15 +3,15 @@ package rawtopic
 import (
 	"context"
 	"fmt"
-	"github.com/google/uuid"
-	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 
+	"github.com/google/uuid"
 	"github.com/ydb-platform/ydb-go-genproto/Ydb_Topic_V1"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopicreader"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopicwriter"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawydb"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
+	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
 type Client struct {

internal/grpcwrapper/rawtopic/rawtopicwriter/streamwriter.go

@@ -3,7 +3,6 @@ package rawtopicwriter
 import (
 	"errors"
 	"fmt"
-	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 	"reflect"
 	"sync"
 	"sync/atomic"
@@ -15,6 +14,7 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopiccommon"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawydb"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
+	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
 var errConcurencyReadDenied = xerrors.Wrap(errors.New("ydb: read from rawtopicwriter in parallel"))
@@ -38,6 +38,7 @@ type StreamWriter struct {
 	sessionID            string
 }
 
+//nolint:funlen
 func (w *StreamWriter) Recv() (ServerMessage, error) {
 	readCnt := atomic.AddInt32(&w.readCounter, 1)
 	defer atomic.AddInt32(&w.readCounter, -1)
@@ -46,22 +47,27 @@ func (w *StreamWriter) Recv() (ServerMessage, error) {
 		return nil, xerrors.WithStackTrace(errConcurencyReadDenied)
 	}
 
-	grpcMsg, err := w.Stream.Recv()
+	grpcMsg, sendErr := w.Stream.Recv()
 	w.readMessagesCount++
-	trace.TopicOnWriterReceiveGRPCMessage(w.Tracer, w.InternalStreamID, w.sessionID, w.readMessagesCount, grpcMsg, err)
-	if err != nil {
-		if !xerrors.IsErrorFromServer(err) {
-			err = xerrors.Transport(err)
+	defer func() {
+		// defer needs for set good session id on first init response before trace the message
+		trace.TopicOnWriterReceiveGRPCMessage(
+			w.Tracer, w.InternalStreamID, w.sessionID, w.readMessagesCount, grpcMsg, sendErr,
+		)
+	}()
+	if sendErr != nil {
+		if !xerrors.IsErrorFromServer(sendErr) {
+			sendErr = xerrors.Transport(sendErr)
 		}
 
 		return nil, xerrors.WithStackTrace(xerrors.Wrap(fmt.Errorf(
 			"ydb: failed to read grpc message from writer stream: %w",
-			err,
+			sendErr,
 		)))
 	}
 
 	var meta rawtopiccommon.ServerMessageMetadata
-	if err = meta.MetaFromStatusAndIssues(grpcMsg); err != nil {
+	if err := meta.MetaFromStatusAndIssues(grpcMsg); err != nil {
 		return nil, err
 	}
 	if !meta.Status.IsSuccess() {
@@ -79,7 +85,7 @@ func (w *StreamWriter) Recv() (ServerMessage, error) {
 	case *Ydb_Topic.StreamWriteMessage_FromServer_WriteResponse:
 		var res WriteResult
 		res.ServerMessageMetadata = meta
-		err = res.fromProto(v.WriteResponse)
+		err := res.fromProto(v.WriteResponse)
 		if err != nil {
 			return nil, err
 		}

internal/topic/topicwriterinternal/writer_reconnector_test.go

@@ -6,7 +6,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 	"io"
 	"math"
 	"sort"
@@ -25,6 +24,7 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xcontext"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xtest"
+	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
 var testCommonEncoders = NewMultiEncoder()

log/topic.go

@@ -2,10 +2,11 @@ package log
 
 import (
 	"context"
+	"time"
+
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Topic"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
-	"time"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/kv"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
@@ -948,17 +949,17 @@ func (s lazyProtoStringifer) String() string {
 				Data     []byte
 				Metadata []*Ydb_Topic.MetadataItem
 			}
-			storage := make([]messDataType, len(data.Messages))
-			for i := range data.Messages {
-				storage[i].Data = data.Messages[i].Data
+			storage := make([]messDataType, len(data.GetMessages()))
+			for i := range data.GetMessages() {
+				storage[i].Data = data.GetMessages()[i].GetData()
 				data.Messages[i] = nil
 
-				storage[i].Metadata = data.Messages[i].MetadataItems
+				storage[i].Metadata = data.GetMessages()[i].GetMetadataItems()
 				data.Messages[i].MetadataItems = nil
 			}
 
 			defer func() {
-				for i := range data.Messages {
+				for i := range data.GetMessages() {
 					data.Messages[i].Data = storage[i].Data
 					data.Messages[i].MetadataItems = storage[i].Metadata
 				}
@@ -967,5 +968,6 @@ func (s lazyProtoStringifer) String() string {
 	}
 
 	res := protojson.MarshalOptions{AllowPartial: true}.Format(s.message)
+
 	return res
 }

tests/integration/topic_read_writer_test.go

@@ -8,7 +8,6 @@ import (
 	"context"
 	"encoding/binary"
 	"errors"
-	"github.com/ydb-platform/ydb-go-sdk/v3/log"
 	"io"
 	"os"
 	"runtime/pprof"
@@ -27,6 +26,7 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/empty"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/version"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xtest"
+	"github.com/ydb-platform/ydb-go-sdk/v3/log"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicsugar"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topictypes"

trace/topic.go

@@ -2,7 +2,8 @@ package trace
 
 import (
 	"context"
-	"google.golang.org/protobuf/proto"
+
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb_Topic"
 )
 
 // tool gtrace used from ./internal/cmd/gtrace
@@ -544,7 +545,7 @@ type (
 		TopicStreamInternalID string
 		SessionID             string
 		MessageNumber         int
-		Message               proto.Message
+		Message               *Ydb_Topic.StreamWriteMessage_FromClient
 		Error                 error
 	}
 
@@ -553,7 +554,7 @@ type (
 		TopicStreamInternalID string
 		SessionID             string
 		MessageNumber         int
-		Message               proto.Message
+		Message               *Ydb_Topic.StreamWriteMessage_FromServer
 		Error                 error
 	}