hubcio commented on code in PR #4497: URL: https://github.com/apache/iggy/pull/4497#discussion_r4322658655
########## foreign/go/client/connection_string.go: ########## @@ -0,0 +1,466 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package client + +import ( + "fmt" + "math/big" + "net" + "strconv" + "strings" + + "github.com/apache/iggy/foreign/go/client/tcp" + ierror "github.com/apache/iggy/foreign/go/errors" +) + +const ( + defaultProtocol = "iggy" + protocolPrefix = "iggy+" + tcpProtocol = "tcp" +) + +// ParseConnectionString parses an Iggy TCP connection string into the TCP +// client options it describes, so the result can be passed to WithTcp: +// +// client.NewIggyClient(client.WithTcp(opts...)) +// +// with opts from client.ParseConnectionString("iggy://iggy:iggy@localhost:8090"). +// +// The accepted shape is: +// +// iggy://[username:password@ | personal_access_token@]host:port[?options] +// iggy+tcp://[username:password@ | personal_access_token@]host:port[?options] +// +// The scheme iggy:// means TCP, the same as the explicit iggy+tcp://. +// Other iggy+<transport> schemes such as iggy+quic:// are rejected, +// because the Go SDK implements the TCP transport only, which covers both +// plain TCP and TLS. Credentials are username:password or a single +// personal access token before the @; either way auto-login is configured, +// matching what the Rust and Node SDKs do with a connection string. The +// host may be a bracketed IPv6 literal, as in [::1]:8090. +// +// Query options follow the grammar the other SDKs share: +// +// tls=true|false enable TLS +// tls_domain=<name> TLS server name indication (SNI) +// tls_ca_file=<path> CA certificate file for verification +// reconnection_retries=<n|unlimited> +// reconnection_interval=<non-zero duration> +// reestablish_after=<duration> +// heartbeat_interval=<non-zero duration> +// nodelay=true|false +// +// Durations take the units the other SDKs share, from nanos to years, +// including the hr, hrs, wk, wks, month and µs spellings, with optional +// fractions and parts run together or spaced, as in 1h30m, 1h 30m or 1 s. +// The reconnection_interval and heartbeat_interval spellings must be +// non-zero and no shorter than one microsecond, because the Rust SDK +// types them as non-zero durations whose wire format carries whole +// microseconds; reestablish_after takes the zero spellings, which Rust +// accepts there. +// The Go client keeps its own reconnect and heartbeat pacing, so those +// three spellings are validated for compatibility and then ignored, and +// a value the other SDKs would reject is still rejected here, so one +// connection string is either valid everywhere or nowhere. The nodelay +// spelling is applied, because the Go client carries a matching +// TCP_NODELAY option; it defaults to true, the Go client default, so +// only nodelay=false ever turns Nagle back on. +func ParseConnectionString(connectionString string) ([]tcp.Option, error) { + protocolParts := strings.Split(connectionString, "://") + if len(protocolParts) != 2 { + return nil, ierror.InvalidConnectionString{} + } + + protocol, err := parseProtocol(protocolParts[0]) + if err != nil { + return nil, err + } + if protocol != tcpProtocol { + // Only a plain transport name is echoed, because a malformed + // scheme may carry credentials, as the C# SDK notes on the same + // check. + if !isASCIILetters(protocol) { + return nil, ierror.InvalidConnectionString{} + } + return nil, fmt.Errorf("%w: unsupported transport %q, Go SDK supports tcp only", + ierror.InvalidConnectionString{}, protocol) + } + + parts := strings.Split(protocolParts[1], "@") + if len(parts) != 2 { + return nil, ierror.InvalidConnectionString{} + } + + credentials, err := parseCredentials(parts[0]) + if err != nil { + return nil, err + } + + serverAndOptions := strings.Split(parts[1], "?") + if len(serverAndOptions) > 2 { + return nil, ierror.InvalidConnectionString{} + } + + // The no-delay default lives here for the no-options path and in + // parseConnectionOptions for queries, so a query that leaves the + // spelling out keeps Nagle off. + var tlsConfig parsedTLS + tlsConfig.noDelay = true + if len(serverAndOptions) == 2 { + tlsConfig, err = parseConnectionOptions(serverAndOptions[1]) + if err != nil { + return nil, err + } + } + + serverAddress := serverAndOptions[0] + if serverAddress == "" { + return nil, ierror.InvalidConnectionString{} + } + host, portStr, err := net.SplitHostPort(serverAddress) + if err != nil || host == "" { + return nil, ierror.InvalidConnectionString{} + } + // Node and Rust accept any u16 port, 0 included; the dial fails later + // instead of the parse. A bad port reports the same error as every other + // parse failure, which is what the Rust SDK returns here. + if _, err := strconv.ParseUint(portStr, 10, 16); err != nil { + return nil, ierror.InvalidConnectionString{} + } + + options := []tcp.Option{tcp.WithServerAddress(serverAddress)} + options = append(options, tcp.WithAutoLogin(credentials)) + if tlsConfig.enabled { + var tlsOpts []tcp.TLSOption + if tlsConfig.domain != "" { + tlsOpts = append(tlsOpts, tcp.WithTLSDomain(tlsConfig.domain)) + } + if tlsConfig.caFile != "" { + tlsOpts = append(tlsOpts, tcp.WithTLSCAFile(tlsConfig.caFile)) + } + options = append(options, tcp.WithTLS(tlsOpts...)) + } + options = append(options, tcp.WithNoDelay(tlsConfig.noDelay)) + return options, nil +} + +// parseProtocol extracts the transport from the scheme, which is either the +// bare default protocol or an explicit iggy+<transport> spelling. +func parseProtocol(scheme string) (string, error) { + if scheme == defaultProtocol { + return tcpProtocol, nil + } + if !strings.HasPrefix(scheme, protocolPrefix) { + return "", ierror.InvalidConnectionString{} + } + return strings.TrimPrefix(scheme, protocolPrefix), nil +} + +// parseCredentials turns the userinfo before the @ into auto-login +// credentials: username:password or a single personal access token. +func parseCredentials(userinfo string) (tcp.Credentials, error) { + username, password, hasSeparator := strings.Cut(userinfo, ":") + if !hasSeparator { + if username == "" { + return tcp.Credentials{}, ierror.InvalidConnectionString{} + } + return tcp.NewPersonalAccessTokenCredentials(username), nil + } + if username == "" || password == "" { + return tcp.Credentials{}, ierror.InvalidConnectionString{} + } + return tcp.NewUsernamePasswordCredentials(username, password), nil +} + +type parsedTLS struct { + enabled bool + domain string + caFile string + noDelay bool +} + +// parseConnectionOptions validates the shared connection-string grammar, +// keeping the tls options and ignoring the reconnect spellings the Go +// client does not take from a connection string. +func parseConnectionOptions(optionsString string) (parsedTLS, error) { + parsed := parsedTLS{noDelay: true} + if optionsString == "" { + return parsed, ierror.InvalidConnectionString{} + } + for _, option := range strings.Split(optionsString, "&") { + name, value, found := strings.Cut(option, "=") + if !found { + return parsed, ierror.InvalidConnectionString{} + } + switch name { + case "tls": + if value != "true" && value != "false" { + return parsed, ierror.InvalidConnectionString{} + } + parsed.enabled = value == "true" + case "tls_domain": + parsed.domain = value + case "tls_ca_file": + parsed.caFile = value + case "reconnection_retries": + if value != "unlimited" { + if _, err := strconv.ParseUint(value, 10, 32); err != nil { + return parsed, ierror.InvalidNumberValue{} + } + } + case "reconnection_interval", "heartbeat_interval": Review Comment: warning: the parser accepts `heartbeat_interval=1s` but keeps the five-second default. pass the parsed interval to `tcp.WithHeartbeatInterval` after a checked conversion to `time.Duration`. ########## foreign/go/client/tcp/tcp_core.go: ########## @@ -450,6 +450,15 @@ func (c *IggyTcpClient) GetConnectionInfo() *iggcon.ConnectionInfo { return &iggcon.ConnectionInfo{Protocol: iggcon.Tcp, ServerAddress: c.links.address()} } +// GetNoDelay reports whether TCP_NODELAY is set on new connections, which +// tests and callers can use to observe the parsed nodelay option before a +// connection exists to read it off. +func (c *IggyTcpClient) GetNoDelay() bool { + c.mtx.Lock() Review Comment: simplification: `GetNoDelay` locks the session mutex to read configuration that never changes after construction. read `c.config.noDelay` directly, as `dialCandidate` does. ########## foreign/go/internal/fakevsr/fakevsr.go: ########## @@ -0,0 +1,296 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package fakevsr provides the fake VSR server the connection-string +// tests drive a full connect-and-login flow against, without a real +// server, and the shared certificate helper the transport tests mint +// their throwaway TLS certificates with. It is an internal package, so +// it never reaches SDK users. +package fakevsr + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/binary" + "encoding/pem" + "io" + "math/big" + "net" + "os" + "path/filepath" + "sync" + "testing" + "time" + + iggcon "github.com/apache/iggy/foreign/go/contracts" + "github.com/apache/iggy/foreign/go/internal/command" + "github.com/apache/iggy/foreign/go/internal/vsr" + "github.com/stretchr/testify/require" +) + +// Frame offsets the fake server reads and writes. They mirror the +// unexported tables of the codec package, which no test can reach. +const ( + FrameOffsetSize = 48 + FrameOffsetCommand = 60 + FrameOffsetRequest = 168 + FrameOffsetOperation = 176 + FrameOffsetSession = 184 + FrameOffsetReserved = 196 + + ReplyFrameOffsetRequest = 200 + ReplyFrameOffsetOperation = 208 + ReplyFrameOffsetStatus = 216 +) + +// Request is one frame the fake server read off the wire. +type Request struct { + Header [vsr.HeaderSize]byte + Payload []byte +} + +// Operation reads the request operation code. +func (r Request) Operation() vsr.Operation { + return vsr.Operation(r.Header[FrameOffsetOperation]) +} + +// Code reads the request command code. +func (r Request) Code() uint32 { + return binary.LittleEndian.Uint32(r.Header[FrameOffsetReserved:]) +} + +// SessionID reads the logical session the request carries. +func (r Request) SessionID() uint64 { + return binary.LittleEndian.Uint64(r.Header[FrameOffsetSession:]) +} + +// ReadRequest reads one complete frame from conn. +func ReadRequest(conn net.Conn) (Request, error) { Review Comment: simplification: this helper and `ClusterMetadataFrame` at line 145 duplicate `readRequest` and `clusterMetadataFrame` in `foreign/go/client/tcp/tcp_testing_test.go`. share these implementations through `fakevsr`, with small wrappers where types differ. ########## foreign/go/internal/fakevsr/fakevsr.go: ########## @@ -0,0 +1,296 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package fakevsr provides the fake VSR server the connection-string +// tests drive a full connect-and-login flow against, without a real +// server, and the shared certificate helper the transport tests mint +// their throwaway TLS certificates with. It is an internal package, so +// it never reaches SDK users. +package fakevsr + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/binary" + "encoding/pem" + "io" + "math/big" + "net" + "os" + "path/filepath" + "sync" + "testing" + "time" + + iggcon "github.com/apache/iggy/foreign/go/contracts" + "github.com/apache/iggy/foreign/go/internal/command" + "github.com/apache/iggy/foreign/go/internal/vsr" + "github.com/stretchr/testify/require" +) + +// Frame offsets the fake server reads and writes. They mirror the +// unexported tables of the codec package, which no test can reach. +const ( + FrameOffsetSize = 48 + FrameOffsetCommand = 60 + FrameOffsetRequest = 168 + FrameOffsetOperation = 176 + FrameOffsetSession = 184 + FrameOffsetReserved = 196 + + ReplyFrameOffsetRequest = 200 + ReplyFrameOffsetOperation = 208 + ReplyFrameOffsetStatus = 216 +) + +// Request is one frame the fake server read off the wire. +type Request struct { + Header [vsr.HeaderSize]byte + Payload []byte +} + +// Operation reads the request operation code. +func (r Request) Operation() vsr.Operation { + return vsr.Operation(r.Header[FrameOffsetOperation]) +} + +// Code reads the request command code. +func (r Request) Code() uint32 { + return binary.LittleEndian.Uint32(r.Header[FrameOffsetReserved:]) +} + +// SessionID reads the logical session the request carries. +func (r Request) SessionID() uint64 { + return binary.LittleEndian.Uint64(r.Header[FrameOffsetSession:]) +} + +// ReadRequest reads one complete frame from conn. +func ReadRequest(conn net.Conn) (Request, error) { + var read Request + if _, err := io.ReadFull(conn, read.Header[:]); err != nil { + return read, err + } + size := binary.LittleEndian.Uint32(read.Header[FrameOffsetSize:]) + if size > vsr.HeaderSize { + read.Payload = make([]byte, int(size)-vsr.HeaderSize) + if _, err := io.ReadFull(conn, read.Payload); err != nil { + return read, err + } + } + return read, nil +} + +// ReplyFrame builds a committed reply carrying body. +func ReplyFrame(operation vsr.Operation, body []byte) []byte { + return StatusReplyFrame(operation, 0, body) +} + +// StatusReplyFrame builds a reply whose header carries a pre-commit denial. +func StatusReplyFrame(operation vsr.Operation, status uint32, body []byte) []byte { + frame := make([]byte, vsr.HeaderSize+len(body)) + binary.LittleEndian.PutUint32(frame[FrameOffsetSize:], uint32(len(frame))) + frame[FrameOffsetCommand] = 8 + binary.LittleEndian.PutUint32(frame[ReplyFrameOffsetStatus:], status) + frame[ReplyFrameOffsetOperation] = byte(operation) + copy(frame[vsr.HeaderSize:], body) + return frame +} + +// ResultSection builds the committed result prefix of a metadata reply. +func ResultSection(codes ...uint32) []byte { + section := binary.LittleEndian.AppendUint32(nil, uint32(len(codes))) + for index, code := range codes { + section = binary.LittleEndian.AppendUint32(section, uint32(index)) + section = binary.LittleEndian.AppendUint32(section, code) + } + return section +} + +// RegisterReplyFrame builds the reply of a successful sign-in. +func RegisterReplyFrame(userID uint32, session uint64) []byte { + body := binary.LittleEndian.AppendUint32(ResultSection(), userID) + body = binary.LittleEndian.AppendUint64(body, session) + body = binary.LittleEndian.AppendUint32(body, vsr.ProtocolVersion) + body = append(body, byte(len("0.11.0-edge.1"))) + body = append(body, "0.11.0-edge.1"...) + return ReplyFrame(vsr.OperationRegister, body) +} + +// BindReplyFrame builds the reply of a successful session bind. +func BindReplyFrame(userID uint32, session uint64) []byte { + registered := RegisterReplyFrame(userID, session) + return ReplyFrame(vsr.OperationNonReplicated, registered[vsr.HeaderSize+len(ResultSection()):]) +} + +// ClusterMetadataFrame builds a cluster roster where the node at +// leaderIndex is the healthy leader. +func ClusterMetadataFrame(t *testing.T, leaderIndex int, addresses ...string) []byte { + t.Helper() + + metadata := iggcon.ClusterMetadata{Name: "test-cluster"} + for index, address := range addresses { + host, port, err := net.SplitHostPort(address) + require.NoError(t, err) + tcpPort, err := net.LookupPort("tcp", port) + require.NoError(t, err) + + role := iggcon.RoleFollower + if index == leaderIndex { + role = iggcon.RoleLeader + } + metadata.Nodes = append(metadata.Nodes, iggcon.ClusterNode{ + ReplicaID: uint8(index), + Name: host, + IP: host, + Role: role, + Status: iggcon.Healthy, + Endpoints: iggcon.TransportEndpoints{Tcp: uint16(tcpPort)}, + }) + } + + body, err := metadata.MarshalBinary() + require.NoError(t, err) + return ReplyFrame(vsr.OperationNonReplicated, body) +} + +// EchoReplyRequest stamps the request id into a reply frame the way the +// server echoes it, so handlers do not repeat the correlation plumbing. An +// eviction frame passes through untouched. +func EchoReplyRequest(answer []byte, read Request) { + if len(answer) < vsr.HeaderSize || answer[FrameOffsetCommand] != 8 { + return + } + copy(answer[ReplyFrameOffsetRequest:ReplyFrameOffsetRequest+8], + read.Header[FrameOffsetRequest:FrameOffsetRequest+8]) +} + +// Server answers frames from a handler and records what it was asked. +type Server struct { + mtx sync.Mutex + requests []Request + listener net.Listener + done chan struct{} Review Comment: simplification: no caller reads `done`. remove the channel field, its allocation, and its deferred close. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
