This is an automated email from the ASF dual-hosted git repository.
spetz pushed a commit to branch go_sdk_vsr
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/go_sdk_vsr by this push:
new 86d62de13 fix
86d62de13 is described below
commit 86d62de133b79b4c9c93c62df6df1421448dcf52
Author: spetz <[email protected]>
AuthorDate: Fri Aug 7 18:23:58 2026 +0200
fix
---
core/sdk/src/vsr.rs | 21 +-
core/server-ng/src/dispatch.rs | 43 ++-
.../binary_response_deserializer.go | 51 +++-
.../vsr_response_deserializer_test.go | 73 +++++
foreign/go/client/tcp/tcp_connect_test.go | 1 +
foreign/go/client/tcp/tcp_core.go | 168 +++++++++--
foreign/go/client/tcp/tcp_core_encode_test.go | 128 +++++++++
foreign/go/client/tcp/tcp_core_review_test.go | 317 +++++++++++++++++++++
foreign/go/client/tcp/tcp_group_polling.go | 54 ++--
foreign/go/client/tcp/tcp_messaging.go | 7 +
foreign/go/client/tcp/tcp_session_management.go | 21 +-
foreign/go/client/tcp/tcp_testing_test.go | 14 +
foreign/go/client/tcp/tcp_topic_cache.go | 9 +-
foreign/go/contracts/client.go | 4 +
foreign/go/contracts/message_header.go | 24 +-
foreign/go/contracts/partitions.go | 11 +-
foreign/go/internal/command/message.go | 162 +++++------
foreign/go/internal/command/message_test.go | 77 +++++
foreign/go/internal/util/leader_aware.go | 40 +--
foreign/go/internal/util/leader_aware_test.go | 5 +
foreign/go/internal/vsr/envelope.go | 10 +-
foreign/go/internal/vsr/header.go | 13 +
foreign/go/internal/vsr/protocol_parity_test.go | 111 +++++++-
foreign/go/internal/vsr/reply.go | 10 +-
foreign/go/internal/vsr/reply_test.go | 4 +-
foreign/go/internal/vsr/session.go | 11 +-
26 files changed, 1170 insertions(+), 219 deletions(-)
diff --git a/core/sdk/src/vsr.rs b/core/sdk/src/vsr.rs
index 9bc5d1cd1..48fd28e31 100644
--- a/core/sdk/src/vsr.rs
+++ b/core/sdk/src/vsr.rs
@@ -90,12 +90,13 @@ pub(crate) fn encode_request_header(
_ => {
let operation = operation_for_code(code);
// NonReplicated ops (ping, reads) bypass server-side dedup --
- // `ClientTable` only tracks request_ids for replicated ops. If
- // they consumed the monotonic counter, the next replicated
- // request would skip an id and the primary's `request_preflight`
- // would see a `RequestGap` and silently drop it. Read the
- // current id without advancing; the server ignores it for
- // NonReplicated.
+ // `ClientTable` only tracks request_ids for replicated ops, and
+ // the table accepts any id above the watermark with no
+ // contiguity requirement (client_table.rs: "There is no
+ // `RequestGap`"), so consuming the counter would not break the
+ // next metadata op. Read the current id without advancing
+ // because the server ignores it for NonReplicated and burning
+ // ids for requests the table never sees buys nothing.
//
// They are also sessionless on the server (routed by transport
// id; protected codes are auth-gated server-side), so send with
@@ -111,9 +112,11 @@ pub(crate) fn encode_request_header(
} else if operation.is_partition() {
// Partition ops replicate in their own per-partition group,
// which is at-least-once with no `ClientTable` dedup -- the
- // metadata table never records their request ids. Consuming
- // the counter here would gap the NEXT metadata op's id and
- // `request_preflight` would silently drop it (`RequestGap`).
+ // metadata table never records their request ids, so there
+ // is nothing for a consumed id to deduplicate against. Every
+ // partition request on a session therefore carries the id
+ // the next metadata op will claim, and a partition-plane
+ // replay is at-least-once.
let session_id =
session.session().ok_or(IggyError::Unauthenticated)?;
(operation, session.current_request_id(), session_id)
} else {
diff --git a/core/server-ng/src/dispatch.rs b/core/server-ng/src/dispatch.rs
index 30f2a05d2..9114d938d 100644
--- a/core/server-ng/src/dispatch.rs
+++ b/core/server-ng/src/dispatch.rs
@@ -2758,10 +2758,27 @@ async fn handle_logout_request<B, MJ, S, SB>(
SB: SuperblockStore + 'static,
{
let Some((vsr_client_id, session)) =
sessions.borrow().get_session(transport_client_id) else {
+ // Logout on an unbound transport: the desired state already holds,
+ // so answer ok. A silent drop would wedge the lockstep SDK on this
+ // connection until its socket read timeout, and the SDK routinely
+ // sends a logout before each re-login.
warn!(
transport_client_id,
- "dropping logout for unbound VSR session"
+ "logout for unbound VSR session; answering ok"
);
+ let commit = current_metadata_commit(shard);
+ let reply = build_empty_reply(request.header(), transport_client_id,
0, commit);
+ if let Err(error) = shard
+ .bus
+ .send_to_client(transport_client_id,
reply.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %error,
+ "failed to send unbound logout reply"
+ );
+ }
return;
};
@@ -2769,7 +2786,29 @@ async fn handle_logout_request<B, MJ, S, SB>(
let commit = match submit_logout_on_owner(shard, vsr_client_id, session,
request_id).await {
Ok(commit) => commit,
Err(error) => {
- warn!(transport_client_id, error = %error, "logout/unregister
failed");
+ // Deny as transient instead of dropping the frame: the submit
+ // usually fails because this replica is not the metadata owner
+ // right now, and the SDK replays a transient rejection.
+ warn!(transport_client_id, error = %error, "logout/unregister
failed; denying transient");
+ let commit = current_metadata_commit(shard);
+ let reply = build_deny_reply(
+ request.header(),
+ vsr_client_id,
+ session,
+ commit,
+ IggyError::TransientNotAccepted.as_code(),
+ );
+ if let Err(send_error) = shard
+ .bus
+ .send_to_client(transport_client_id,
reply.into_generic().into_frozen())
+ .await
+ {
+ warn!(
+ transport_client_id,
+ error = %send_error,
+ "failed to send logout deny reply"
+ );
+ }
return;
}
};
diff --git a/foreign/go/binary_serialization/binary_response_deserializer.go
b/foreign/go/binary_serialization/binary_response_deserializer.go
index 1f1af561d..35cf03ac9 100644
--- a/foreign/go/binary_serialization/binary_response_deserializer.go
+++ b/foreign/go/binary_serialization/binary_response_deserializer.go
@@ -129,6 +129,13 @@ func DeserializeToStream(payload []byte, position int)
(iggcon.Stream, int, erro
}, totalSize, nil
}
+// pollBatchHeaderLength covers [partition_id u32][current_offset u64][count
u32].
+const pollBatchHeaderLength = 16
+
+// DeserializeFetchMessagesResponse decodes a poll reply. A truncated body is
+// a decode error rather than a shorter batch: silently dropping the tail
+// would let a consumer that commits CurrentOffset skip messages it never saw.
+// The returned messages alias the reply buffer; a retained message pins it.
func DeserializeFetchMessagesResponse(payload []byte, compression
iggcon.IggyMessageCompression) (*iggcon.PolledMessage, error) {
if len(payload) == 0 {
return &iggcon.PolledMessage{
@@ -139,31 +146,44 @@ func DeserializeFetchMessagesResponse(payload []byte,
compression iggcon.IggyMes
}
length := len(payload)
+ if length < pollBatchHeaderLength {
+ return nil, fmt.Errorf("poll response: %d bytes is short of the
batch header", length)
+ }
partitionId := binary.LittleEndian.Uint32(payload[0:4])
currentOffset := binary.LittleEndian.Uint64(payload[4:12])
messagesCount := binary.LittleEndian.Uint32(payload[12:16])
- position := 16
- var messages = make([]iggcon.IggyMessage, 0)
+ position := pollBatchHeaderLength
+
+ // The declared count is server-controlled; the allocation hint is
capped
+ // by what the body could possibly hold.
+ maxMessages := (length - pollBatchHeaderLength) /
iggcon.MessageHeaderSize
+ if int(messagesCount) < maxMessages {
+ maxMessages = int(messagesCount)
+ }
+ messages := make([]iggcon.IggyMessage, 0, maxMessages)
for position < length {
- if position+iggcon.MessageHeaderSize >= length {
- // body needs to be at least 1 byte
- break
+ if position+iggcon.MessageHeaderSize > length {
+ return nil, fmt.Errorf("poll response: truncated
message header at byte %d", position)
}
header, err := iggcon.MessageHeaderFromBytes(payload[position :
position+iggcon.MessageHeaderSize])
if err != nil {
return nil, err
}
position += iggcon.MessageHeaderSize
- payload_end := position + int(header.PayloadLength)
- if int(payload_end) > length {
- break
+ if uint64(header.PayloadLength) > uint64(length-position) {
+ return nil, fmt.Errorf(
+ "poll response: message payload of %d bytes
overruns the body", header.PayloadLength)
}
- payloadSlice := payload[position:payload_end]
- position = int(payload_end)
+ payloadSlice := payload[position :
position+int(header.PayloadLength)]
+ position += int(header.PayloadLength)
- var user_headers []byte = nil
+ if uint64(header.UserHeaderLength) > uint64(length-position) {
+ return nil, fmt.Errorf(
+ "poll response: user headers of %d bytes
overrun the body", header.UserHeaderLength)
+ }
+ var userHeaders []byte
if header.UserHeaderLength > 0 {
- user_headers = payload[position :
position+int(header.UserHeaderLength)]
+ userHeaders = payload[position :
position+int(header.UserHeaderLength)]
}
position += int(header.UserHeaderLength)
@@ -181,9 +201,14 @@ func DeserializeFetchMessagesResponse(payload []byte,
compression iggcon.IggyMes
messages = append(messages, iggcon.IggyMessage{
Header: *header,
Payload: payloadSlice,
- UserHeaders: user_headers,
+ UserHeaders: userHeaders,
})
}
+ if uint32(len(messages)) != messagesCount {
+ return nil, fmt.Errorf(
+ "poll response: %d decoded messages do not match the
declared %d",
+ len(messages), messagesCount)
+ }
// !TODO: Add message offset ordering
return &iggcon.PolledMessage{
diff --git a/foreign/go/binary_serialization/vsr_response_deserializer_test.go
b/foreign/go/binary_serialization/vsr_response_deserializer_test.go
index d029aeca4..7a43ab14c 100644
--- a/foreign/go/binary_serialization/vsr_response_deserializer_test.go
+++ b/foreign/go/binary_serialization/vsr_response_deserializer_test.go
@@ -167,6 +167,79 @@ func
TestDeserializeConsumerGroupAssignment_DoesNotAllocateOnABogusCount(t *test
assert.Error(t, err)
}
+// pollPayload builds a poll reply body carrying the given messages.
+func pollPayload(t *testing.T, partitionId uint32, payloads ...[]byte) []byte {
+ t.Helper()
+
+ body := binary.LittleEndian.AppendUint32(nil, partitionId)
+ body = binary.LittleEndian.AppendUint64(body, 42)
+ body = binary.LittleEndian.AppendUint32(body, uint32(len(payloads)))
+ for _, payload := range payloads {
+ message, err := iggcon.NewIggyMessage(payload)
+ require.NoError(t, err)
+ headerBytes, err := message.Header.AppendBinary(nil)
+ require.NoError(t, err)
+ body = append(body, headerBytes...)
+ body = append(body, message.Payload...)
+ }
+ return body
+}
+
+func TestDeserializeFetchMessagesResponse_DecodesABatch(t *testing.T) {
+ payload := pollPayload(t, 3, []byte("first"), []byte("second"))
+
+ polled, err := DeserializeFetchMessagesResponse(payload,
iggcon.MESSAGE_COMPRESSION_NONE)
+ require.NoError(t, err)
+ assert.Equal(t, uint32(3), polled.PartitionId)
+ assert.Equal(t, uint64(42), polled.CurrentOffset)
+ require.Len(t, polled.Messages, 2)
+ assert.Equal(t, []byte("first"), polled.Messages[0].Payload)
+ assert.Equal(t, []byte("second"), polled.Messages[1].Payload)
+}
+
+func TestDeserializeFetchMessagesResponse_TreatsAnEmptyPayloadAsAnEmptyBatch(t
*testing.T) {
+ polled, err := DeserializeFetchMessagesResponse(nil,
iggcon.MESSAGE_COMPRESSION_NONE)
+ require.NoError(t, err)
+ assert.Empty(t, polled.Messages)
+}
+
+func TestDeserializeFetchMessagesResponse_RejectsEveryTruncation(t *testing.T)
{
+ payload := pollPayload(t, 1, []byte("first"), []byte("second"))
+
+ for length := 1; length < len(payload); length++ {
+ _, err := DeserializeFetchMessagesResponse(
+ payload[:length], iggcon.MESSAGE_COMPRESSION_NONE)
+ assert.Error(t, err,
+ "a reply truncated to %d bytes must not decode as a
shorter batch", length)
+ }
+}
+
+func
TestDeserializeFetchMessagesResponse_RejectsACountAboveTheDecodedMessages(t
*testing.T) {
+ payload := pollPayload(t, 1, []byte("only"))
+ binary.LittleEndian.PutUint32(payload[12:16], 5)
+
+ _, err := DeserializeFetchMessagesResponse(payload,
iggcon.MESSAGE_COMPRESSION_NONE)
+ assert.Error(t, err,
+ "a declared count the body does not satisfy must surface, not
silently shrink")
+}
+
+func TestDeserializeFetchMessagesResponse_RejectsOverrunningUserHeaders(t
*testing.T) {
+ message, err := iggcon.NewIggyMessage([]byte("payload"))
+ require.NoError(t, err)
+ message.Header.UserHeaderLength = 64
+
+ body := binary.LittleEndian.AppendUint32(nil, 1)
+ body = binary.LittleEndian.AppendUint64(body, 0)
+ body = binary.LittleEndian.AppendUint32(body, 1)
+ headerBytes, err := message.Header.AppendBinary(nil)
+ require.NoError(t, err)
+ body = append(body, headerBytes...)
+ body = append(body, message.Payload...)
+
+ _, err = DeserializeFetchMessagesResponse(body,
iggcon.MESSAGE_COMPRESSION_NONE)
+ assert.Error(t, err, "user headers past the body must not panic or
pass")
+}
+
func TestDeserializeToTopic_PinsTheFieldOrderOfTheWireLayout(t *testing.T) {
const nameOffset = 50
name := "orders"
diff --git a/foreign/go/client/tcp/tcp_connect_test.go
b/foreign/go/client/tcp/tcp_connect_test.go
index 5e090040d..5a812b725 100644
--- a/foreign/go/client/tcp/tcp_connect_test.go
+++ b/foreign/go/client/tcp/tcp_connect_test.go
@@ -127,6 +127,7 @@ func listenVSR(t *testing.T, wrap func(net.Conn) net.Conn,
if answer == nil {
return
}
+ echoReplyRequest(answer, read)
if _, err := conn.Write(answer); err !=
nil {
return
}
diff --git a/foreign/go/client/tcp/tcp_core.go
b/foreign/go/client/tcp/tcp_core.go
index 526b864e3..e8ec0e950 100644
--- a/foreign/go/client/tcp/tcp_core.go
+++ b/foreign/go/client/tcp/tcp_core.go
@@ -18,6 +18,7 @@
package tcp
import (
+ "bufio"
"context"
"crypto/tls"
"crypto/x509"
@@ -28,6 +29,7 @@ import (
"log/slog"
"net"
"os"
+ "slices"
"sync"
"time"
@@ -51,8 +53,19 @@ func GetDefaultOptions() Options {
}
type IggyTcpClient struct {
- conn net.Conn
- mtx sync.Mutex
+ conn net.Conn
+ // reader buffers reads off conn, so a reply costs one syscall instead
of
+ // one for the header and one for the body; guarded by c.mtx.
+ reader *bufio.Reader
+ mtx sync.Mutex
+ // registerMtx single-flights the sign-in transaction: BeginRegister and
+ // Bind live in different c.mtx critical sections, and two interleaved
+ // sign-ins would commit one Register whose session is never bound.
+ registerMtx sync.Mutex
+ // closed unblocks a replay wait when Close is called, which cannot go
+ // through c.mtx because the replay loop holds it.
+ closed chan struct{}
+ closeOnce sync.Once
config config
logger *slog.Logger
MessageCompression iggcon.IggyMessageCompression
@@ -69,6 +82,9 @@ type IggyTcpClient struct {
// skipAutoLoginOnce suppresses the next automatic sign-in so a replayed
// login is not preempted by one the reconnect issues; guarded by c.mtx.
skipAutoLoginOnce bool
+ // loggedOut records an explicit sign-out, so a reconnect's automatic
+ // sign-in does not silently reverse it; guarded by c.mtx.
+ loggedOut bool
// groups caches the consumer-group assignments this client polls with.
groups groupAssignmentCache
// topics caches what a send needs to resolve a partition locally.
@@ -98,7 +114,9 @@ func defaultTcpClientConfig() config {
tls: defaultTLSConfig(),
autoLogin: AutoLogin{},
reconnection: defaultTcpClientReconnectionConfig(),
- noDelay: false,
+ // The lockstep request model stalls a multi-segment frame
behind
+ // Nagle's algorithm, so coalescing is off by default.
+ noDelay: true,
}
}
@@ -221,6 +239,15 @@ func WithTLSValidateCertificate(validate bool) TLSOption {
}
}
+// WithNoDelay controls TCP_NODELAY. It defaults to true because the lockstep
+// request model stalls behind Nagle's algorithm; pass false to re-enable
+// segment coalescing for bandwidth-bound workloads.
+func WithNoDelay(noDelay bool) Option {
+ return func(opts *Options) {
+ opts.config.noDelay = noDelay
+ }
+}
+
// NewIggyTcpClient creates a new Iggy TCP client with the given options.
// warning: don't use this function directly, use iggycli.NewIggyClient with
iggycli.WithTcp instead.
func NewIggyTcpClient(logger *slog.Logger, options ...Option) *IggyTcpClient {
@@ -245,6 +272,7 @@ func NewIggyTcpClient(logger *slog.Logger, options
...Option) *IggyTcpClient {
leaderRedirectionState: iggcon.LeaderRedirectionState{},
currentServerAddress: opts.config.serverAddress,
session: vsr.NewSession(),
+ closed: make(chan struct{}),
}
}
@@ -268,14 +296,12 @@ const (
failoverCheckInterval = 2 * time.Second
)
-// requestPrologue is the zeroed space a frame reserves ahead of its payload.
-// The header is stamped into it once the payload length is known.
-var requestPrologue [vsr.HeaderSize]byte
-
-// requestBufPool reuses wire-payload buffers across RPCs.
+// requestBufPool reuses wire-payload buffers across RPCs. A fresh buffer
+// already holds the header plus room for a small payload, so a metadata
+// command does not reallocate on its first byte.
var requestBufPool = sync.Pool{
New: func() any {
- b := make([]byte, 0, 256)
+ b := make([]byte, 0, vsr.HeaderSize+512)
return &b
},
}
@@ -285,7 +311,10 @@ func acquireRequestBuf() *[]byte {
}
func releaseRequestBuf(bp *[]byte) {
- const maxPooled = 64 * 1024
+ // Producer batches routinely grow to megabytes and reusing them is the
+ // point of the pool; the ceiling only stops a pathological frame from
+ // pinning memory for the process lifetime.
+ const maxPooled = 4 * 1024 * 1024
if cap(*bp) > maxPooled {
return
}
@@ -302,8 +331,12 @@ func (c *IggyTcpClient) read(expectedSize int) (int,
[]byte, error) {
return n, buffer, nil
}
-// readInto reads exactly len(buf) bytes from the connection into buf.
+// readInto reads exactly len(buf) bytes from the connection into buf, through
+// the buffered reader when the connect flow installed one.
func (c *IggyTcpClient) readInto(buf []byte) (int, error) {
+ if c.reader != nil {
+ return io.ReadFull(c.reader, buf)
+ }
var totalRead int
expected := len(buf)
for totalRead < expected {
@@ -342,10 +375,14 @@ func (c *IggyTcpClient) do(ctx context.Context, cmd
command.Command) ([]byte, er
defer releaseRequestBuf(bp)
frame, err := appendCommandFrame(*bp, cmd)
+ if frame != nil {
+ // Keep the grown buffer even when the encode failed, so the
pool gets
+ // it back at its new capacity.
+ *bp = frame
+ }
if err != nil {
return nil, err
}
- *bp = frame
return c.exchange(ctx, uint32(cmd.Code()), frame)
}
@@ -360,12 +397,20 @@ func (c *IggyTcpClient) SendBinaryRequest(ctx
context.Context, code uint32, payl
bp := acquireRequestBuf()
defer releaseRequestBuf(bp)
- frame := append(append((*bp)[:0], requestPrologue[:]...), payload...)
+ frame := append(reserveHeader(*bp), payload...)
*bp = frame
return c.exchange(ctx, code, frame)
}
+// reserveHeader returns buf truncated to exactly the header prologue, growing
+// it as needed. The prologue bytes are not zeroed here: EncodeRequestHeader
+// clears the full header when the frame is stamped, and zeroing twice would
+// write four extra cache lines per request.
+func reserveHeader(buf []byte) []byte {
+ return slices.Grow(buf[:0], vsr.HeaderSize)[:vsr.HeaderSize]
+}
+
func isSessionControlCode(code uint32) bool {
switch code {
case uint32(command.LoginUserCode),
@@ -386,16 +431,18 @@ func isRegisterCode(code uint32) bool {
}
// appendCommandFrame reserves the request prologue in buf and appends the
-// encoded command payload after it, so the frame is a single allocation the
-// header can be stamped into once the payload length is known.
+// encoded command payload after it. A command implementing
+// encoding.BinaryAppender encodes straight into the buffer, keeping the frame
+// a single allocation; the MarshalBinary fallback pays one extra body
+// allocation and copy.
func appendCommandFrame(buf []byte, cmd command.Command) ([]byte, error) {
- buf = append(buf[:0], requestPrologue[:]...)
+ buf = reserveHeader(buf)
if appender, ok := cmd.(encoding.BinaryAppender); ok {
return appender.AppendBinary(buf)
}
body, err := cmd.MarshalBinary()
if err != nil {
- return nil, err
+ return buf, err
}
return append(buf, body...), nil
}
@@ -405,6 +452,14 @@ func appendCommandFrame(buf []byte, cmd command.Command)
([]byte, error) {
// the stack, and re-entering it has no depth bound.
type connectScoped struct{}
+// localPreconditionError marks a request that failed before its frame was
+// written. The connection is healthy, so exchange must not tear it down and
+// re-dial over what is purely local state.
+type localPreconditionError struct{ err error }
+
+func (e *localPreconditionError) Error() string { return e.err.Error() }
+func (e *localPreconditionError) Unwrap() error { return e.err }
+
// exchange runs one request to completion, reconnecting and replaying it when
// the failure is one a fresh connection recovers from.
func (c *IggyTcpClient) exchange(ctx context.Context, code uint32, frame
[]byte) ([]byte, error) {
@@ -412,6 +467,10 @@ func (c *IggyTcpClient) exchange(ctx context.Context, code
uint32, frame []byte)
if err == nil || !isReconnectable(err) {
return response, err
}
+ var precondition *localPreconditionError
+ if errors.As(err, &precondition) {
+ return nil, err
+ }
if ctx.Value(connectScoped{}) != nil {
return nil, err
}
@@ -428,6 +487,14 @@ func (c *IggyTcpClient) exchange(ctx context.Context, code
uint32, frame []byte)
if !c.config.autoLogin.enabled && !login {
return nil, err
}
+ c.mtx.Lock()
+ loggedOut := c.loggedOut
+ c.mtx.Unlock()
+ if loggedOut && !login {
+ // An explicit sign-out stays signed out: the reconnect's
automatic
+ // sign-in would silently reverse it.
+ return nil, err
+ }
if !canReplay(code, err) {
c.logger.Warn("Not replaying a replicated request with an
unknown outcome.",
slog.Int("code", int(code)), slog.Any("error", err))
@@ -588,8 +655,10 @@ func (c *IggyTcpClient) attempt(
if !stamped {
// Stamp once per session: the header consumes a request id,
and a
// replay must carry the same one for the server to deduplicate
it.
+ // A stamp failure is local and pre-write, so it is marked as
such:
+ // the connection is healthy and must not be torn down for it.
if err := vsr.StampRequestHeader(c.session, code, frame); err
!= nil {
- return nil, false, err
+ return nil, false, &localPreconditionError{err}
}
stamped = true
}
@@ -642,26 +711,44 @@ func (c *IggyTcpClient) exchangeLocked(
c.logger.Debug("Sending a TCP request",
slog.Int("frame_length", len(frame)), slog.Int("code",
int(code)))
if _, err := c.write(frame); err != nil {
+ // Normalized like the read failures below, so callers
can match
+ // a dropped connection with one sentinel in both
directions.
+ c.logger.Error("Failed to write the request frame",
+ slog.Int("code", int(code)), slog.Any("error",
err))
c.invalidateConnLocked()
- return nil, err
+ return nil, ierror.ErrDisconnected
}
body, err := c.readReplyLocked(code)
if err != nil {
return nil, err
}
+ if vsr.PeekCommand(&c.respHeader) == vsr.FrameReply {
+ // A reply must answer the request in flight. An
unexpected echo
+ // means the stream is delivering some other request's
answer, and
+ // every later reply would pair off by one.
+ expected := vsr.StampedRequestID(frame)
+ if echoed := vsr.ReadReplyRequestID(&c.respHeader);
echoed != expected {
+ c.logger.Error("The reply answers a different
request",
+ slog.Uint64("expected_request",
expected),
+ slog.Uint64("echoed_request", echoed))
+ c.invalidateConnLocked()
+ return nil, ierror.ErrDisconnected
+ }
+ }
response, err := vsr.DecodeReply(&c.respHeader, body)
switch {
case errors.Is(err, ierror.ErrTransientNotCommitted) &&
time.Now().Before(readDeadline):
// The outcome is unknown, so only a replay of the same
request id
- // on this session is safe. The server's client table
answers from
- // its reply cache if the request did commit.
- if waitErr := waitBeforeReplay(ctx, readDeadline);
waitErr != nil {
+ // on this session is safe. On the metadata plane the
client table
+ // answers a committed request from its reply cache;
the partition
+ // plane keeps no client table, so its replay is
at-least-once.
+ if waitErr := c.waitBeforeReplay(ctx, readDeadline);
waitErr != nil {
return nil, waitErr
}
case errors.Is(err, ierror.ErrTransientNotAccepted) &&
time.Now().Before(transientDeadline):
- if waitErr := waitBeforeReplay(ctx, transientDeadline);
waitErr != nil {
+ if waitErr := c.waitBeforeReplay(ctx,
transientDeadline); waitErr != nil {
return nil, waitErr
}
default:
@@ -723,8 +810,11 @@ func (c *IggyTcpClient) handleReplyFailureLocked(err
error) {
}
// waitBeforeReplay pauses before resending a transiently rejected request,
-// never past the deadline and never past the caller's cancellation.
-func waitBeforeReplay(ctx context.Context, deadline time.Time) error {
+// never past the deadline, the caller's cancellation, or a client shutdown.
+// The shutdown channel matters because this wait runs with c.mtx held, which
+// is the lock Close needs; without it Close would block for the rest of the
+// request budget.
+func (c *IggyTcpClient) waitBeforeReplay(ctx context.Context, deadline
time.Time) error {
interval := replayInterval
if remaining := time.Until(deadline); remaining < interval {
interval = remaining
@@ -738,6 +828,8 @@ func waitBeforeReplay(ctx context.Context, deadline
time.Time) error {
select {
case <-ctx.Done():
return ctx.Err()
+ case <-c.closed:
+ return ierror.ErrClientShutdown
case <-timer.C:
return nil
}
@@ -750,7 +842,7 @@ func (c *IggyTcpClient) invalidateConnLocked() {
c.sessionState = iggcon.SessionStateUnauthenticated
c.session.Reset()
c.groups.clear()
- c.topics.clear()
+ c.topics.clearCounts()
}
// closeConnLocked closes and drops the current connection.
@@ -760,6 +852,7 @@ func (c *IggyTcpClient) closeConnLocked() error {
}
err := c.conn.Close()
c.conn = nil
+ c.reader = nil
return err
}
@@ -881,6 +974,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
c.mtx.Lock()
c.conn = conn
+ c.reader = bufio.NewReaderSize(conn, 64*1024)
c.transportState = iggcon.TransportStateConnected
c.connectedAt = time.Now()
// The server fence does not survive the old socket, so the new
connection
@@ -940,6 +1034,16 @@ func (c *IggyTcpClient) establishSession(ctx
context.Context, skipAutoLogin bool
return err
}
if redirect {
+ if skipAutoLogin {
+ // The suppression belongs to the sign-in replay, not
to this
+ // connection attempt. Connect already consumed the
flag, so it is
+ // re-armed for the post-redirect Connect; otherwise
that nested
+ // Connect signs in automatically and the replayed
login then
+ // commits a second Register.
+ c.mtx.Lock()
+ c.skipAutoLoginOnce = true
+ c.mtx.Unlock()
+ }
return c.Connect(ctx)
}
@@ -1018,7 +1122,7 @@ func (c *IggyTcpClient) disconnect() error {
c.sessionState = iggcon.SessionStateUnauthenticated
c.session.Reset()
c.groups.clear()
- c.topics.clear()
+ c.topics.clearCounts()
err := c.closeConnLocked()
@@ -1028,6 +1132,14 @@ func (c *IggyTcpClient) disconnect() error {
}
func (c *IggyTcpClient) shutdown() error {
+ // Unblock any in-flight replay wait before taking the lock it holds,
+ // otherwise this call queues behind the rest of that request's budget.
+ c.closeOnce.Do(func() {
+ if c.closed != nil {
+ close(c.closed)
+ }
+ })
+
c.mtx.Lock()
defer c.mtx.Unlock()
@@ -1043,7 +1155,7 @@ func (c *IggyTcpClient) shutdown() error {
c.sessionState = iggcon.SessionStateUnauthenticated
c.session.Reset()
c.groups.clear()
- c.topics.clear()
+ c.topics.clearCounts()
c.logger.Info("Iggy TCP client has been shutdown.",
slog.String("client_address", c.clientAddress))
// TODO push shutdown event
return err
diff --git a/foreign/go/client/tcp/tcp_core_encode_test.go
b/foreign/go/client/tcp/tcp_core_encode_test.go
new file mode 100644
index 000000000..3ba348830
--- /dev/null
+++ b/foreign/go/client/tcp/tcp_core_encode_test.go
@@ -0,0 +1,128 @@
+// 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 tcp
+
+import (
+ "bytes"
+ "errors"
+ "testing"
+
+ 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/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// marshalOnlyCmd implements command.Command without AppendBinary; it forces
+// appendCommandFrame down the MarshalBinary fallback branch.
+type marshalOnlyCmd struct {
+ body []byte
+ code command.Code
+ wantErr error
+}
+
+func (f *marshalOnlyCmd) Code() command.Code { return f.code }
+func (f *marshalOnlyCmd) MarshalBinary() ([]byte, error) {
+ if f.wantErr != nil {
+ return nil, f.wantErr
+ }
+ return bytes.Clone(f.body), nil
+}
+
+func testPollCmd(t *testing.T) *command.PollMessages {
+ t.Helper()
+ partitionId := uint32(7)
+ return &command.PollMessages{
+ Consumer: iggcon.NewSingleConsumer(numericIdentifier(t, 42)),
+ StreamId: numericIdentifier(t, 1),
+ TopicId: numericIdentifier(t, 2),
+ PartitionId: &partitionId,
+ Strategy: iggcon.FirstPollingStrategy(),
+ Count: 100,
+ AutoCommit: true,
+ }
+}
+
+func TestAppendCommandFrame_AppenderPathMatchesTheFallback(t *testing.T) {
+ cmd := testPollCmd(t)
+ body, err := cmd.MarshalBinary()
+ require.NoError(t, err)
+ fallback := &marshalOnlyCmd{body: body, code: cmd.Code()}
+
+ want, err := appendCommandFrame(nil, fallback)
+ require.NoError(t, err)
+ got, err := appendCommandFrame(nil, cmd)
+ require.NoError(t, err)
+
+ assert.Equal(t, want[vsr.HeaderSize:], got[vsr.HeaderSize:],
+ "the fast path diverges from the MarshalBinary fallback")
+ assert.Len(t, got, vsr.HeaderSize+len(body),
+ "the frame is the reserved prologue followed by the payload")
+}
+
+func TestAppendCommandFrame_FallbackPropagatesTheMarshalError(t *testing.T) {
+ sentinel := errors.New("marshal failed")
+ cmd := &marshalOnlyCmd{code: command.Code(1), wantErr: sentinel}
+
+ frame, err := appendCommandFrame(make([]byte, 0, 4), cmd)
+ assert.ErrorIs(t, err, sentinel)
+ assert.NotNil(t, frame, "the grown buffer survives the error so the
pool keeps it")
+}
+
+func TestAppendCommandFrame_GrowsAnUndersizedBuffer(t *testing.T) {
+ cmd := testPollCmd(t)
+ want, err := appendCommandFrame(make([]byte, 0, 4096), cmd)
+ require.NoError(t, err)
+
+ got, err := appendCommandFrame(make([]byte, 0, 4), cmd)
+ require.NoError(t, err)
+ assert.Equal(t, want[vsr.HeaderSize:], got[vsr.HeaderSize:])
+}
+
+func TestRequestBufPool_AcquireReleaseRoundTrip(t *testing.T) {
+ bp := acquireRequestBuf()
+ require.NotNil(t, bp)
+ require.NotNil(t, *bp)
+ assert.GreaterOrEqual(t, cap(*bp), vsr.HeaderSize,
+ "a fresh buffer already holds the header prologue")
+
+ *bp = append(*bp, []byte("hello")...)
+ releaseRequestBuf(bp)
+
+ next := acquireRequestBuf()
+ require.NotNil(t, next)
+ assert.Empty(t, *next, "a pooled buffer comes back empty")
+ releaseRequestBuf(next)
+}
+
+func TestRequestBufPool_KeepsABatchSizedBuffer(t *testing.T) {
+ batch := make([]byte, 0, 1<<20)
+ releaseRequestBuf(&batch)
+ // sync.Pool gives no delivery guarantee; what must hold is that the
+ // release accepted the megabyte buffer instead of dropping it on the
+ // floor, which the truncation observes.
+ assert.Empty(t, batch)
+}
+
+func TestRequestBufPool_DropsAPathologicalBuffer(t *testing.T) {
+ huge := make([]byte, 0, 8<<20)
+ huge = append(huge, 1)
+ releaseRequestBuf(&huge)
+ assert.Len(t, huge, 1, "an over-ceiling buffer is not truncated, it is
dropped")
+}
diff --git a/foreign/go/client/tcp/tcp_core_review_test.go
b/foreign/go/client/tcp/tcp_core_review_test.go
new file mode 100644
index 000000000..15261d62f
--- /dev/null
+++ b/foreign/go/client/tcp/tcp_core_review_test.go
@@ -0,0 +1,317 @@
+// 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 tcp
+
+import (
+ "context"
+ "encoding/binary"
+ "net"
+ "sync"
+ "testing"
+ "time"
+
+ iggcon "github.com/apache/iggy/foreign/go/contracts"
+ ierror "github.com/apache/iggy/foreign/go/errors"
+ "github.com/apache/iggy/foreign/go/internal/command"
+ "github.com/apache/iggy/foreign/go/internal/vsr"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// newUnboundPipeClient is newPipeClient without the session bind, for tests
+// that need the unauthenticated state.
+func newUnboundPipeClient(t *testing.T) (*IggyTcpClient, net.Conn) {
+ t.Helper()
+
+ serverConn, clientConn := net.Pipe()
+ client := newTestClient(t, clientConn)
+ t.Cleanup(func() {
+ _ = clientConn.Close()
+ _ = serverConn.Close()
+ })
+ return client, serverConn
+}
+
+func TestExchange_DoesNotReconnectOnALocalStampFailure(t *testing.T) {
+ client, serverConn := newUnboundPipeClient(t)
+ client.config.autoLogin =
NewAutoLogin(NewUsernamePasswordCredentials("iggy", "iggy"))
+ server := serve(serverConn, func(_ int, _ request) []byte {
+ return replyFrame(vsr.OperationCreateStream, resultSection())
+ })
+
+ // The session is unbound, so stamping a replicated request fails
locally
+ // before any byte reaches the wire.
+ _, err := client.do(context.Background(), &command.CreateStream{Name:
"orders"})
+
+ assert.ErrorIs(t, err, ierror.ErrUnauthenticated)
+ assert.Equal(t, iggcon.TransportStateConnected, client.transportState,
+ "a healthy connection survives a purely local failure")
+ assert.Empty(t, server.recorded(), "nothing reached the wire")
+}
+
+func TestLogoutUser_StaysSignedOutThroughTheReconnectPath(t *testing.T) {
+ var server *testListener
+ server = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ return clusterMetadataFrame(t, 0, server.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, 128)
+ case read.operation() == vsr.OperationLogout:
+ return replyFrame(vsr.OperationLogout, nil)
+ default:
+ // The server refuses the signed-out session, which is
the shape
+ // that previously drove reconnect plus automatic
re-login.
+ return evictionFrame(vsr.EvictionNoSession, 0, 0)
+ }
+ })
+
+ client := newDialingClient(t, server.address(),
+ WithAutoLogin(NewUsernamePasswordCredentials("iggy", "iggy")))
+ require.NoError(t, client.Connect(context.Background()))
+ require.NoError(t, client.LogoutUser(context.Background()))
+
+ err := client.Ping(context.Background())
+ assert.ErrorIs(t, err, ierror.ErrUnauthenticated,
+ "the caller hears about the missing session instead of being
re-logged in")
+
+ registers := 0
+ for _, read := range server.recorded() {
+ if read.operation() == vsr.OperationRegister {
+ registers++
+ }
+ }
+ assert.Equal(t, 1, registers, "an explicit sign-out is not reversed by
an automatic sign-in")
+ assert.Equal(t, 1, server.connections(), "no reconnect was attempted")
+}
+
+func TestLoginUser_ClearsTheExplicitLogoutSuppression(t *testing.T) {
+ var server *testListener
+ server = listenVSR(t, nil, func(connection, _ int, read request) []byte
{
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ return clusterMetadataFrame(t, 0, server.address())
+ case read.operation() == vsr.OperationRegister:
+ return registerReplyFrame(7, uint64(128+connection))
+ case read.operation() == vsr.OperationLogout:
+ return replyFrame(vsr.OperationLogout, nil)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+
+ client := newDialingClient(t, server.address(),
+ WithAutoLogin(NewUsernamePasswordCredentials("iggy", "iggy")))
+ require.NoError(t, client.Connect(context.Background()))
+ require.NoError(t, client.LogoutUser(context.Background()))
+
+ _, err := client.LoginUser(context.Background(), "iggy", "iggy")
+ require.NoError(t, err)
+
+ client.mtx.Lock()
+ loggedOut := client.loggedOut
+ client.mtx.Unlock()
+ assert.False(t, loggedOut, "an explicit sign-in lifts the suppression")
+ require.NoError(t, client.Ping(context.Background()))
+}
+
+func TestConnect_RedirectDuringAReplayedLoginDoesNotAutoSignIn(t *testing.T) {
+ var leader *testListener
+ leader = listenVSR(t, nil, singleNodeHandler(t, func() string { return
leader.address() }))
+ follower := listenVSR(t, nil, func(_, _ int, read request) []byte {
+ if read.code() == uint32(command.GetClusterMetadataCode) {
+ return clusterMetadataFrame(t, 1, "127.0.0.1:1",
leader.address())
+ }
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ })
+
+ client := newDialingClient(t, follower.address(),
+ WithAutoLogin(NewUsernamePasswordCredentials("iggy", "iggy")))
+ // The state a replayed login leaves behind before its reconnect.
+ client.skipAutoLoginOnce = true
+
+ require.NoError(t, client.Connect(context.Background()))
+
+ for _, read := range leader.recorded() {
+ assert.NotEqual(t, vsr.OperationRegister, read.operation(),
+ "the redirected Connect must keep suppressing the
automatic sign-in")
+ }
+ assert.False(t, client.skipAutoLoginOnce, "the suppression is consumed
exactly once")
+}
+
+func TestClose_InterruptsAnInFlightReplayWait(t *testing.T) {
+ client, serverConn := newPipeClient(t)
+ serve(serverConn, func(_ int, _ request) []byte {
+ return statusReplyFrame(vsr.OperationCreateStream,
+ uint32(ierror.TransientNotCommittedCode), nil)
+ })
+
+ done := make(chan error, 1)
+ go func() {
+ _, err := client.do(context.Background(),
&command.CreateStream{Name: "orders"})
+ done <- err
+ }()
+ time.Sleep(50 * time.Millisecond)
+
+ closed := time.Now()
+ require.NoError(t, client.Close())
+ assert.Less(t, time.Since(closed), 5*time.Second,
+ "Close must not wait out the request budget")
+
+ select {
+ case err := <-done:
+ assert.ErrorIs(t, err, ierror.ErrClientShutdown)
+ case <-time.After(5 * time.Second):
+ t.Fatal("the in-flight request never returned after Close")
+ }
+}
+
+func TestExchange_DropsTheConnectionOnAMismatchedReplyEcho(t *testing.T) {
+ client, serverConn := newPipeClient(t)
+ client.config.reconnection.enabled = false
+ go func() {
+ read, err := readRequest(serverConn)
+ if err != nil {
+ return
+ }
+ answer := replyFrame(vsr.OperationCreateStream, resultSection())
+ binary.LittleEndian.PutUint64(answer[replyFrameOffsetRequest:],
+ read.requestID()+7)
+ _, _ = serverConn.Write(answer)
+ }()
+
+ _, err := client.do(context.Background(), &command.CreateStream{Name:
"orders"})
+ assert.ErrorIs(t, err, ierror.ErrDisconnected,
+ "a reply for some other request means the stream is desynced")
+ assert.Equal(t, iggcon.TransportStateDisconnected,
client.transportState)
+}
+
+func TestExchange_NormalizesAWriteFailure(t *testing.T) {
+ client, serverConn := newPipeClient(t)
+ client.config.reconnection.enabled = false
+ _ = serverConn.Close()
+
+ _, err := client.SendBinaryRequest(context.Background(),
uint32(command.PingCode), nil)
+ assert.ErrorIs(t, err, ierror.ErrDisconnected,
+ "a write failure classifies like a read failure")
+ assert.Equal(t, iggcon.TransportStateDisconnected,
client.transportState)
+}
+
+func TestLoginUser_ConcurrentSignInsAreSingleFlight(t *testing.T) {
+ var server *testListener
+ sessions := uint64(128)
+ var sessionMtx sync.Mutex
+ server = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ switch {
+ case read.code() == uint32(command.GetClusterMetadataCode):
+ return clusterMetadataFrame(t, 0, server.address())
+ case read.operation() == vsr.OperationRegister:
+ sessionMtx.Lock()
+ sessions++
+ session := sessions
+ sessionMtx.Unlock()
+ return registerReplyFrame(7, session)
+ case read.operation() == vsr.OperationLogout:
+ return replyFrame(vsr.OperationLogout, nil)
+ default:
+ return replyFrame(vsr.OperationNonReplicated, nil)
+ }
+ })
+
+ client := newDialingClient(t, server.address())
+ require.NoError(t, client.Connect(context.Background()))
+
+ var group sync.WaitGroup
+ errs := make([]error, 2)
+ for i := range errs {
+ group.Add(1)
+ go func(index int) {
+ defer group.Done()
+ _, errs[index] = client.LoginUser(context.Background(),
"iggy", "iggy")
+ }(i)
+ }
+ group.Wait()
+
+ assert.NoError(t, errs[0])
+ assert.NoError(t, errs[1])
+ assert.True(t, client.session.Bound(),
+ "the surviving sign-in leaves one bound session, not an
orphaned register")
+}
+
+func TestLoginUser_DropsTheConnectionWhenTheBindFails(t *testing.T) {
+ var server *testListener
+ server = listenVSR(t, nil, func(_, _ int, read request) []byte {
+ if read.code() == uint32(command.GetClusterMetadataCode) {
+ return clusterMetadataFrame(t, 0, server.address())
+ }
+ // A zero session fence is not bindable.
+ return registerReplyFrame(7, 0)
+ })
+
+ client := newDialingClient(t, server.address())
+ require.NoError(t, client.Connect(context.Background()))
+
+ _, err := client.LoginUser(context.Background(), "iggy", "iggy")
+ require.Error(t, err)
+ assert.Equal(t, iggcon.TransportStateDisconnected,
client.transportState,
+ "the server committed a Register this client cannot adopt; the
connection is unusable")
+}
+
+func TestTopicCache_ClearCountsKeepsTheBalancedCursors(t *testing.T) {
+ cache := topicCache{}
+ key := topicKey{stream: "s", topic: "t"}
+ cache.setPartitionsCount(key, 4)
+ require.Equal(t, uint32(0), cache.nextBalanced(key, 4))
+ require.Equal(t, uint32(1), cache.nextBalanced(key, 4))
+
+ cache.clearCounts()
+
+ _, cached := cache.partitionsCount(key)
+ assert.False(t, cached, "the count must be reread after a reconnect")
+ assert.Equal(t, uint32(2), cache.nextBalanced(key, 4),
+ "the fairness cursor survives, so a fleet-wide reconnect does
not stampede partition 0")
+}
+
+func TestSendMessages_InvalidatesTheCountWhenThePartitionVanished(t
*testing.T) {
+ client, serverConn := newPipeClient(t)
+ serve(serverConn, func(_ int, read request) []byte {
+ if read.operation() == vsr.OperationSendMessages {
+ return statusReplyFrame(vsr.OperationSendMessages,
+ uint32(ierror.PartitionNotFoundCode), nil)
+ }
+ return replyFrame(vsr.OperationNonReplicated,
topicDetailsBody(t, 4))
+ })
+
+ streamId := numericIdentifier(t, 1)
+ topicId := numericIdentifier(t, 1)
+ message, err := iggcon.NewIggyMessage([]byte("payload"))
+ require.NoError(t, err)
+
+ _, err = client.SendMessages(context.Background(),
+ streamId, topicId, iggcon.None(), []iggcon.IggyMessage{message})
+ require.ErrorIs(t, err, ierror.ErrPartitionNotFound)
+
+ _, cached := client.topics.partitionsCount(newTopicKey(streamId,
topicId))
+ assert.False(t, cached,
+ "the topic was likely recreated smaller; the count must be
reread")
+}
+
+func TestGetDefaultOptions_DisablesNagle(t *testing.T) {
+ assert.True(t, GetDefaultOptions().config.noDelay,
+ "a lockstep client must not stall frames behind Nagle's
algorithm")
+}
diff --git a/foreign/go/client/tcp/tcp_group_polling.go
b/foreign/go/client/tcp/tcp_group_polling.go
index 246b6728c..92b8e5321 100644
--- a/foreign/go/client/tcp/tcp_group_polling.go
+++ b/foreign/go/client/tcp/tcp_group_polling.go
@@ -89,14 +89,37 @@ func (c *groupAssignmentCache) put(key groupKey, assignment
groupAssignment) {
c.entries[key] = &assignment
}
-// advance moves the round-robin cursor of a cached entry past the partition
-// that was just polled.
-func (c *groupAssignmentCache) advance(key groupKey, cursor int) {
+// nextPartition returns the partition the next poll targets and advances the
+// round-robin cursor, both under one lock. A separate read and advance would
+// let two concurrent polls target the same partition and skip another.
+func (c *groupAssignmentCache) nextPartition(key groupKey) (uint32, bool) {
c.mtx.Lock()
defer c.mtx.Unlock()
- if entry, ok := c.entries[key]; ok {
- entry.cursor = cursor
+ entry, ok := c.entries[key]
+ if !ok || len(entry.partitions) == 0 {
+ return 0, false
}
+ cursor := entry.cursor % len(entry.partitions)
+ entry.cursor = (cursor + 1) % len(entry.partitions)
+ return entry.partitions[cursor], true
+}
+
+// putSynced stores a fresh sync result, carrying the round-robin cursor
+// forward when the generation is unchanged, all under one lock so a
+// concurrent poll's advance is never overwritten with a stale value.
+func (c *groupAssignmentCache) putSynced(key groupKey, assignment
groupAssignment) {
+ c.mtx.Lock()
+ defer c.mtx.Unlock()
+ if c.entries == nil {
+ c.entries = make(map[groupKey]*groupAssignment)
+ }
+ if current, ok := c.entries[key]; ok && current.generation ==
assignment.generation {
+ // A same-generation refresh is not a rebalance: the member
still owns
+ // the same partitions, so the round-robin position carries over
+ // instead of resetting and starving the partitions behind it.
+ assignment.cursor = current.cursor
+ }
+ c.entries[key] = &assignment
}
func (c *groupAssignmentCache) drop(key groupKey) {
@@ -176,9 +199,11 @@ func (c *IggyTcpClient) pollGroup(
}, nil
}
- cursor := assignment.cursor % len(assignment.partitions)
- partitionId := assignment.partitions[cursor]
- c.groups.advance(key, (cursor+1)%len(assignment.partitions))
+ partitionId, owned := c.groups.nextPartition(key)
+ if !owned {
+ // The entry vanished between the sync and this read;
re-sync.
+ continue
+ }
polled, err := c.pollPartition(
ctx, streamId, topicId, consumer, strategy, count,
autoCommit, &partitionId)
@@ -191,8 +216,9 @@ func (c *IggyTcpClient) pollGroup(
}
// The server marks a stale assignment by answering an empty
batch on
- // the resync sentinel partition.
- if polled.PartitionId == iggcon.ResyncRequiredPartition &&
polled.MessageCount == 0 {
+ // the resync sentinel partition. The gate reads the decoded
batch, not
+ // the server-declared count, which is the same signal Rust
checks.
+ if polled.PartitionId == iggcon.ResyncRequiredPartition &&
len(polled.Messages) == 0 {
c.groups.drop(key)
continue
}
@@ -246,15 +272,9 @@ func (c *IggyTcpClient) ensureAssignment(
partitions: synced.Partitions,
fetchedAt: time.Now(),
}
- if current, ok := c.groups.get(key); ok && current.generation ==
synced.Generation {
- // A same-generation refresh is not a rebalance: the member
still owns
- // the same partitions, so the round-robin position carries over
- // instead of resetting and starving the partitions behind it.
- assignment.cursor = current.cursor
- }
c.logger.Debug("Synced the consumer group assignment",
slog.Uint64("generation", assignment.generation),
slog.Int("partitions", len(assignment.partitions)))
- c.groups.put(key, assignment)
+ c.groups.putSynced(key, assignment)
return assignment, nil
}
diff --git a/foreign/go/client/tcp/tcp_messaging.go
b/foreign/go/client/tcp/tcp_messaging.go
index 6e54b2f72..95441b90a 100644
--- a/foreign/go/client/tcp/tcp_messaging.go
+++ b/foreign/go/client/tcp/tcp_messaging.go
@@ -19,6 +19,7 @@ package tcp
import (
"context"
+ "errors"
"log/slog"
binaryserialization
"github.com/apache/iggy/foreign/go/binary_serialization"
@@ -59,6 +60,12 @@ func (c *IggyTcpClient) SendMessages(
Messages: messages,
})
if err != nil {
+ if errors.Is(err, ierror.ErrPartitionNotFound) {
+ // The cached count pointed this send at a partition
the server
+ // does not have, so the topic was likely recreated
smaller.
+ // Nothing else invalidates a count another client
changed.
+
c.topics.invalidatePartitionsCount(newTopicKey(streamId, topicId))
+ }
return nil, err
}
diff --git a/foreign/go/client/tcp/tcp_session_management.go
b/foreign/go/client/tcp/tcp_session_management.go
index 0102f38b3..3ca38e432 100644
--- a/foreign/go/client/tcp/tcp_session_management.go
+++ b/foreign/go/client/tcp/tcp_session_management.go
@@ -52,6 +52,14 @@ func (c *IggyTcpClient) LoginWithPersonalAccessToken(ctx
context.Context, token
// the existing session untouched, and a connection that dies mid-attempt is
// already reset by invalidateConnLocked.
func (c *IggyTcpClient) register(ctx context.Context, code uint32, body
[]byte) (*iggcon.IdentityInfo, error) {
+ // One sign-in at a time. BeginRegister runs inside the exchange lock
but
+ // Bind runs after it, so two interleaved sign-ins would let the second
+ // BeginRegister reset the identity the first is about to bind: one
+ // committed Register would be orphaned in the server's client table and
+ // the losing caller would see ErrSessionAlreadyBound.
+ c.registerMtx.Lock()
+ defer c.registerMtx.Unlock()
+
c.logger.Info("Iggy client is signing in...",
slog.String("client_address", c.clientAddress))
if err := c.endBoundSession(ctx); err != nil {
@@ -60,7 +68,7 @@ func (c *IggyTcpClient) register(ctx context.Context, code
uint32, body []byte)
bp := acquireRequestBuf()
defer releaseRequestBuf(bp)
- frame := append(append((*bp)[:0], requestPrologue[:]...), body...)
+ frame := append(reserveHeader(*bp), body...)
*bp = frame
response, err := c.exchange(ctx, code, frame)
@@ -77,6 +85,12 @@ func (c *IggyTcpClient) register(ctx context.Context, code
uint32, body []byte)
err = c.session.Bind(registered.Session)
if err == nil {
c.sessionState = iggcon.SessionStateAuthenticated
+ c.loggedOut = false
+ } else {
+ // The server committed a Register this client failed to adopt,
so the
+ // connection carries a session the local state does not track.
It is
+ // unusable; drop it like any other terminal session failure.
+ c.invalidateConnLocked()
}
c.mtx.Unlock()
if err != nil {
@@ -108,8 +122,11 @@ func (c *IggyTcpClient) LogoutUser(ctx context.Context)
error {
c.mtx.Lock()
c.sessionState = iggcon.SessionStateUnauthenticated
c.session.Reset()
+ // The sign-out is caller intent: it suppresses the automatic sign-in on
+ // the reconnect path until the caller explicitly signs in again.
+ c.loggedOut = true
c.groups.clear()
- c.topics.clear()
+ c.topics.clearCounts()
c.mtx.Unlock()
return nil
}
diff --git a/foreign/go/client/tcp/tcp_testing_test.go
b/foreign/go/client/tcp/tcp_testing_test.go
index 0f9d0ec78..b63671b0e 100644
--- a/foreign/go/client/tcp/tcp_testing_test.go
+++ b/foreign/go/client/tcp/tcp_testing_test.go
@@ -42,6 +42,7 @@ const (
frameOffsetSession = 192
frameOffsetReserved = 204
+ replyFrameOffsetRequest = 200
replyFrameOffsetOperation = 208
replyFrameOffsetStatus = 224
@@ -153,6 +154,17 @@ func clusterMetadataFrame(t *testing.T, leaderIndex int,
addresses ...string) []
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])
+}
+
// readRequest reads one complete frame from conn.
func readRequest(conn net.Conn) (request, error) {
var read request
@@ -195,6 +207,7 @@ func serve(conn net.Conn, handler func(index int, read
request) []byte) *fakeSer
if answer == nil {
continue
}
+ echoReplyRequest(answer, read)
if _, err := conn.Write(answer); err != nil {
return
}
@@ -237,5 +250,6 @@ func newTestClient(t *testing.T, conn net.Conn)
*IggyTcpClient {
logger: slog.New(slog.DiscardHandler),
session: vsr.NewSession(),
config: defaultTcpClientConfig(),
+ closed: make(chan struct{}),
}
}
diff --git a/foreign/go/client/tcp/tcp_topic_cache.go
b/foreign/go/client/tcp/tcp_topic_cache.go
index f830b007c..dd0508235 100644
--- a/foreign/go/client/tcp/tcp_topic_cache.go
+++ b/foreign/go/client/tcp/tcp_topic_cache.go
@@ -112,10 +112,13 @@ func (c *topicCache) nextBalanced(key topicKey,
partitionsCount uint32) uint32 {
return partition
}
-// clear forgets topic state so a new session rereads current metadata.
-func (c *topicCache) clear() {
+// clearCounts forgets the partition counts so a new session rereads current
+// metadata. Balanced cursors survive on purpose: they are client-side
+// fairness state with no dependency on cluster metadata, and resetting them
+// on a reconnect would point every producer's first post-failover batch at
+// partition 0 at once.
+func (c *topicCache) clearCounts() {
c.mtx.Lock()
defer c.mtx.Unlock()
clear(c.partitionsCounts)
- clear(c.balancedCursors)
}
diff --git a/foreign/go/contracts/client.go b/foreign/go/contracts/client.go
index db4ee1792..6bcf71cd5 100644
--- a/foreign/go/contracts/client.go
+++ b/foreign/go/contracts/client.go
@@ -120,6 +120,10 @@ type Client interface {
// example after an explicit LeaveConsumerGroup. JoinConsumerGroup
// restores membership.
// - err == nil with a real partition id: messages were read.
+ //
+ // The returned message payloads and user headers alias the reply
buffer,
+ // so retaining one message pins the whole reply; copy the bytes out
when
+ // they outlive the poll.
PollMessages(
ctx context.Context,
streamId Identifier,
diff --git a/foreign/go/contracts/message_header.go
b/foreign/go/contracts/message_header.go
index 7167f18d8..8b6129280 100644
--- a/foreign/go/contracts/message_header.go
+++ b/foreign/go/contracts/message_header.go
@@ -74,17 +74,17 @@ func MessageHeaderFromBytes(data []byte) (*MessageHeader,
error) {
}
func (mh *MessageHeader) ToBytes() []byte {
- bytes := make([]byte, 0, MessageHeaderSize)
-
- bytes = binary.LittleEndian.AppendUint64(bytes, mh.Checksum)
- idBytes := mh.Id[:]
- bytes = append(bytes, idBytes...)
- bytes = binary.LittleEndian.AppendUint64(bytes, mh.Offset)
- bytes = binary.LittleEndian.AppendUint64(bytes, mh.Timestamp)
- bytes = binary.LittleEndian.AppendUint64(bytes, mh.OriginTimestamp)
- bytes = binary.LittleEndian.AppendUint32(bytes, mh.UserHeaderLength)
- bytes = binary.LittleEndian.AppendUint32(bytes, mh.PayloadLength)
- bytes = binary.LittleEndian.AppendUint64(bytes, mh.Reserved)
-
+ bytes, _ := mh.AppendBinary(make([]byte, 0, MessageHeaderSize))
return bytes
}
+
+func (mh *MessageHeader) AppendBinary(b []byte) ([]byte, error) {
+ b = binary.LittleEndian.AppendUint64(b, mh.Checksum)
+ b = append(b, mh.Id[:]...)
+ b = binary.LittleEndian.AppendUint64(b, mh.Offset)
+ b = binary.LittleEndian.AppendUint64(b, mh.Timestamp)
+ b = binary.LittleEndian.AppendUint64(b, mh.OriginTimestamp)
+ b = binary.LittleEndian.AppendUint32(b, mh.UserHeaderLength)
+ b = binary.LittleEndian.AppendUint32(b, mh.PayloadLength)
+ return binary.LittleEndian.AppendUint64(b, mh.Reserved), nil
+}
diff --git a/foreign/go/contracts/partitions.go
b/foreign/go/contracts/partitions.go
index c7043d3c0..309ea4412 100644
--- a/foreign/go/contracts/partitions.go
+++ b/foreign/go/contracts/partitions.go
@@ -120,9 +120,10 @@ func EntityIdGuid(value uuid.UUID) Partitioning {
}
func (p Partitioning) MarshalBinary() ([]byte, error) {
- bytes := make([]byte, 2+p.Length)
- bytes[0] = byte(p.Kind)
- bytes[1] = byte(p.Length)
- copy(bytes[2:], p.Value)
- return bytes, nil
+ return p.AppendBinary(make([]byte, 0, 2+p.Length))
+}
+
+func (p Partitioning) AppendBinary(b []byte) ([]byte, error) {
+ b = append(b, byte(p.Kind), byte(p.Length))
+ return append(b, p.Value...), nil
}
diff --git a/foreign/go/internal/command/message.go
b/foreign/go/internal/command/message.go
index 7e34a11d4..ef9bcc9a8 100644
--- a/foreign/go/internal/command/message.go
+++ b/foreign/go/internal/command/message.go
@@ -46,109 +46,91 @@ func (s *SendMessages) Code() Code {
return SendMessagesCode
}
+// zeroIndex is the blank per-message index entry reserved ahead of the
+// message section and filled in as messages are appended.
+var zeroIndex [indexSize]byte
+
func (s *SendMessages) MarshalBinary() ([]byte, error) {
- for i, message := range s.Messages {
- switch s.Compression {
- case iggcon.MESSAGE_COMPRESSION_S2:
- if len(message.Payload) < 32 {
- break
- }
- s.Messages[i].Payload = s2.Encode(nil, message.Payload)
- message.Header.PayloadLength =
uint32(len(message.Payload))
- case iggcon.MESSAGE_COMPRESSION_S2_BETTER:
- if len(message.Payload) < 32 {
- break
- }
- s.Messages[i].Payload = s2.EncodeBetter(nil,
message.Payload)
- message.Header.PayloadLength =
uint32(len(message.Payload))
- case iggcon.MESSAGE_COMPRESSION_S2_BEST:
- if len(message.Payload) < 32 {
- break
- }
- s.Messages[i].Payload = s2.EncodeBest(nil,
message.Payload)
- message.Header.PayloadLength =
uint32(len(message.Payload))
- }
- }
+ return s.AppendBinary(nil)
+}
- streamIdBytes, err := s.StreamId.MarshalBinary()
- if err != nil {
- return nil, err
+// AppendBinary encodes the batch straight into b: [metadata_length u32]
+// [stream id][topic id][partitioning][messages_count u32], the per-message
+// index section, then each message as header, payload, user headers.
+func (s *SendMessages) AppendBinary(b []byte) ([]byte, error) {
+ s.compressPayloads()
+
+ metadataStart := len(b)
+ b = binary.LittleEndian.AppendUint32(b, 0)
+ var err error
+ if b, err = s.StreamId.AppendBinary(b); err != nil {
+ return b, err
}
- topicIdBytes, err := s.TopicId.MarshalBinary()
- if err != nil {
- return nil, err
+ if b, err = s.TopicId.AppendBinary(b); err != nil {
+ return b, err
}
- partitioningBytes, err := s.Partitioning.MarshalBinary()
- if err != nil {
- return nil, err
+ if b, err = s.Partitioning.AppendBinary(b); err != nil {
+ return b, err
}
- metadataLenFieldSize := 4 // uint32
- messageCount := len(s.Messages)
- messagesCountFieldSize := 4 // uint32
- metadataLen := len(streamIdBytes) +
- len(topicIdBytes) +
- len(partitioningBytes) +
- messagesCountFieldSize
- indexesSize := messageCount * indexSize
- messageBytesCount := calculateMessageBytesCount(s.Messages)
- totalSize := metadataLenFieldSize +
- len(streamIdBytes) +
- len(topicIdBytes) +
- len(partitioningBytes) +
- messagesCountFieldSize +
- indexesSize +
- messageBytesCount
-
- bytes := make([]byte, totalSize)
-
- position := 0
-
- //metadata
- binary.LittleEndian.PutUint32(bytes[:4], uint32(metadataLen))
- position = 4
- //ids
- copy(bytes[position:position+len(streamIdBytes)], streamIdBytes)
- position += len(streamIdBytes)
- copy(bytes[position:position+len(topicIdBytes)], topicIdBytes)
- position += len(topicIdBytes)
-
- //partitioning
- copy(bytes[position:position+len(partitioningBytes)], partitioningBytes)
- position += len(partitioningBytes)
- binary.LittleEndian.PutUint32(bytes[position:position+4],
uint32(messageCount))
- position += 4
-
- currentIndexPosition := position
- for i := 0; i < indexesSize; i++ {
- bytes[position+i] = 0
+ b = binary.LittleEndian.AppendUint32(b, uint32(len(s.Messages)))
+ metadataLength := len(b) - metadataStart - 4
+ binary.LittleEndian.PutUint32(b[metadataStart:], uint32(metadataLength))
+
+ indexesStart := len(b)
+ for range s.Messages {
+ b = append(b, zeroIndex[:]...)
}
- position += indexesSize
msgSize := uint32(0)
- for _, message := range s.Messages {
- copy(bytes[position:position+iggcon.MessageHeaderSize],
message.Header.ToBytes())
-
copy(bytes[position+iggcon.MessageHeaderSize:position+iggcon.MessageHeaderSize+int(message.Header.PayloadLength)],
message.Payload)
- position += iggcon.MessageHeaderSize +
int(message.Header.PayloadLength)
-
copy(bytes[position:position+int(message.Header.UserHeaderLength)],
message.UserHeaders)
- position += int(message.Header.UserHeaderLength)
-
- msgSize += iggcon.MessageHeaderSize +
message.Header.PayloadLength + message.Header.UserHeaderLength
-
-
binary.LittleEndian.PutUint32(bytes[currentIndexPosition:currentIndexPosition+4],
0)
-
binary.LittleEndian.PutUint32(bytes[currentIndexPosition+4:currentIndexPosition+8],
uint32(msgSize))
-
binary.LittleEndian.PutUint32(bytes[currentIndexPosition+8:currentIndexPosition+12],
0)
- currentIndexPosition += indexSize
+ for i := range s.Messages {
+ message := &s.Messages[i]
+ // The header lengths and the appended slices must agree, or
every
+ // message boundary after a mismatch mis-frames; deriving both
from
+ // the same slice makes the disagreement impossible.
+ message.Header.PayloadLength = uint32(len(message.Payload))
+ message.Header.UserHeaderLength =
uint32(len(message.UserHeaders))
+ if b, err = message.Header.AppendBinary(b); err != nil {
+ return b, err
+ }
+ b = append(b, message.Payload...)
+ b = append(b, message.UserHeaders...)
+
+ msgSize += iggcon.MessageHeaderSize +
+ message.Header.PayloadLength +
message.Header.UserHeaderLength
+ binary.LittleEndian.PutUint32(b[indexesStart+i*indexSize+4:],
msgSize)
}
- return bytes, nil
+ return b, nil
}
-func calculateMessageBytesCount(messages []iggcon.IggyMessage) int {
- count := 0
- for _, msg := range messages {
- count += iggcon.MessageHeaderSize + len(msg.Payload) +
len(msg.UserHeaders)
+// compressPayloads compresses each payload in place. The header length is
+// updated through the slice index: writing it to a range copy would leave the
+// wire header claiming the uncompressed length, and the encoder would then
+// mis-frame every message that follows.
+func (s *SendMessages) compressPayloads() {
+ switch s.Compression {
+ case iggcon.MESSAGE_COMPRESSION_S2,
+ iggcon.MESSAGE_COMPRESSION_S2_BETTER,
+ iggcon.MESSAGE_COMPRESSION_S2_BEST:
+ default:
+ return
+ }
+
+ for i := range s.Messages {
+ payload := s.Messages[i].Payload
+ if len(payload) < 32 {
+ continue
+ }
+ switch s.Compression {
+ case iggcon.MESSAGE_COMPRESSION_S2:
+ s.Messages[i].Payload = s2.Encode(nil, payload)
+ case iggcon.MESSAGE_COMPRESSION_S2_BETTER:
+ s.Messages[i].Payload = s2.EncodeBetter(nil, payload)
+ case iggcon.MESSAGE_COMPRESSION_S2_BEST:
+ s.Messages[i].Payload = s2.EncodeBest(nil, payload)
+ }
+ s.Messages[i].Header.PayloadLength =
uint32(len(s.Messages[i].Payload))
}
- return count
}
type PollMessages struct {
diff --git a/foreign/go/internal/command/message_test.go
b/foreign/go/internal/command/message_test.go
index 305f5274b..1fbe43ca7 100644
--- a/foreign/go/internal/command/message_test.go
+++ b/foreign/go/internal/command/message_test.go
@@ -23,6 +23,7 @@ import (
"github.com/apache/iggy/foreign/go/contracts"
"github.com/google/uuid"
+ "github.com/klauspost/compress/s2"
)
func TestSerialize_TcpFetchMessagesRequest(t *testing.T) {
@@ -134,6 +135,82 @@ func TestSerialize_SendMessagesRequest(t *testing.T) {
}
}
+func TestSerialize_SendMessagesCompressesThePayloadCoherently(t *testing.T) {
+ // A compressible payload above the 32-byte floor.
+ payload := bytes.Repeat([]byte("abcdefgh"), 32)
+ message, err := iggcon.NewIggyMessage(payload)
+ if err != nil {
+ t.Fatal(err)
+ }
+ streamId, _ := iggcon.NewIdentifier(uint32(1))
+ topicId, _ := iggcon.NewIdentifier(uint32(1))
+ request := SendMessages{
+ StreamId: streamId,
+ TopicId: topicId,
+ Partitioning: iggcon.PartitionId(0),
+ Messages: []iggcon.IggyMessage{message},
+ Compression: iggcon.MESSAGE_COMPRESSION_S2,
+ }
+
+ serialized, err := request.MarshalBinary()
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ compressed := request.Messages[0]
+ if got := int(compressed.Header.PayloadLength); got !=
len(compressed.Payload) {
+ t.Fatalf("header claims %d payload bytes, the slice holds %d:
the wire would mis-frame",
+ got, len(compressed.Payload))
+ }
+ if len(compressed.Payload) >= len(payload) {
+ t.Fatalf("the payload did not compress: %d >= %d",
len(compressed.Payload), len(payload))
+ }
+
+ // The message section must be framed by the compressed length: header,
+ // then exactly PayloadLength payload bytes, and nothing after.
+ messageStart := len(serialized) - int(compressed.Header.PayloadLength)
- iggcon.MessageHeaderSize
+ header, err := iggcon.MessageHeaderFromBytes(
+ serialized[messageStart :
messageStart+iggcon.MessageHeaderSize])
+ if err != nil {
+ t.Fatal(err)
+ }
+ if header.PayloadLength != compressed.Header.PayloadLength {
+ t.Fatalf("wire header claims %d, in-memory header %d",
+ header.PayloadLength, compressed.Header.PayloadLength)
+ }
+
+ decoded, err := s2.Decode(nil,
serialized[messageStart+iggcon.MessageHeaderSize:])
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(decoded, payload) {
+ t.Fatal("the compressed payload does not round-trip")
+ }
+}
+
+func TestSerialize_SendMessagesSkipsCompressionBelowTheFloor(t *testing.T) {
+ message, err := iggcon.NewIggyMessage([]byte("short"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ streamId, _ := iggcon.NewIdentifier(uint32(1))
+ topicId, _ := iggcon.NewIdentifier(uint32(1))
+ request := SendMessages{
+ StreamId: streamId,
+ TopicId: topicId,
+ Partitioning: iggcon.PartitionId(0),
+ Messages: []iggcon.IggyMessage{message},
+ Compression: iggcon.MESSAGE_COMPRESSION_S2,
+ }
+
+ if _, err := request.MarshalBinary(); err != nil {
+ t.Fatal(err)
+ }
+ if !bytes.Equal(request.Messages[0].Payload, []byte("short")) {
+ t.Fatal("a payload under 32 bytes must pass through
uncompressed")
+ }
+}
+
func createDefaultMessageHeaders() []iggcon.HeaderEntry {
return []iggcon.HeaderEntry{
{Key: iggcon.HeaderKey{Kind: iggcon.String, Value:
[]byte("HeaderKey1")}, Value: iggcon.HeaderValue{Kind: iggcon.String, Value:
[]byte("Value 1")}},
diff --git a/foreign/go/internal/util/leader_aware.go
b/foreign/go/internal/util/leader_aware.go
index 4be742362..ce4e86c83 100644
--- a/foreign/go/internal/util/leader_aware.go
+++ b/foreign/go/internal/util/leader_aware.go
@@ -149,30 +149,32 @@ func clusterNodeAddress(node *iggcon.ClusterNode,
transport iggcon.Protocol) (st
return net.JoinHostPort(node.IP, strconv.Itoa(int(port))), nil
}
-// isSameAddress returns true if two addresses refer to the same endpoint.
+// isSameAddress reports whether two addresses refer to the same endpoint.
+// The comparison is lexical after normalization, with an IP-literal fast path
+// for spelling differences like ::1 versus 0:0:0:0:0:0:0:1. It deliberately
+// never resolves names: both sides come from the same cluster-metadata roster
+// or config, and this runs on the failover path where a resolver lookup would
+// block with neither a context nor a deadline to interrupt it.
func isSameAddress(addr1, addr2 string) bool {
- a1 := parseAddress(addr1)
- a2 := parseAddress(addr2)
-
- if a1 != nil && a2 != nil {
- return a1.IP.Equal(a2.IP) && a1.Port == a2.Port
+ host1, port1 := splitAddress(normalizeAddress(addr1))
+ host2, port2 := splitAddress(normalizeAddress(addr2))
+ if port1 != port2 {
+ return false
}
-
- return normalizeAddress(addr1) == normalizeAddress(addr2)
+ if ip1, ip2 := net.ParseIP(host1), net.ParseIP(host2); ip1 != nil &&
ip2 != nil {
+ return ip1.Equal(ip2)
+ }
+ return host1 == host2
}
-// parseAddress attempts to parse an address into a *net.TCPAddr.
-func parseAddress(addr string) *net.TCPAddr {
- // Try direct parse
- if ta, err := net.ResolveTCPAddr("tcp", addr); err == nil {
- return ta
- }
- // Normalize then try again
- normalized := normalizeAddress(addr)
- if ta, err := net.ResolveTCPAddr("tcp", normalized); err == nil {
- return ta
+// splitAddress splits host:port, treating an unparsable address as a bare
+// host so the comparison degrades to a string match instead of failing.
+func splitAddress(addr string) (string, string) {
+ host, port, err := net.SplitHostPort(addr)
+ if err != nil {
+ return addr, ""
}
- return nil
+ return host, port
}
// normalizeAddress canonicalizes address strings for fallback comparison.
diff --git a/foreign/go/internal/util/leader_aware_test.go
b/foreign/go/internal/util/leader_aware_test.go
index 5407bd792..7624b1623 100644
--- a/foreign/go/internal/util/leader_aware_test.go
+++ b/foreign/go/internal/util/leader_aware_test.go
@@ -139,6 +139,11 @@ func TestIsSameAddress(t *testing.T) {
{"localhost", "localhost:8090", "127.0.0.1:8090", true},
{"different port", "127.0.0.1:8090", "127.0.0.1:8091", false},
{"different ip", "192.168.1.1:8090", "127.0.0.1:8090", false},
+ {"ipv6 spellings", "[::1]:8090", "[0:0:0:0:0:0:0:1]:8090",
true},
+ {"same hostname", "IGGY.local:8090", "iggy.local:8090", true},
+ // Hostnames compare lexically: this path runs during failover,
where
+ // a resolver lookup could block with no deadline.
+ {"hostname versus its ip", "iggy.local:8090", "10.0.0.1:8090",
false},
}
for _, tc := range cases {
diff --git a/foreign/go/internal/vsr/envelope.go
b/foreign/go/internal/vsr/envelope.go
index a56a93874..690b5d910 100644
--- a/foreign/go/internal/vsr/envelope.go
+++ b/foreign/go/internal/vsr/envelope.go
@@ -52,8 +52,7 @@ func StampRequestHeader(session *Session, code uint32, frame
[]byte) error {
}
// Namespace derivation can fail on a malformed payload. Run it before
- // taking a request id so a local failure never burns one, which would
gap
- // the sequence and make the primary drop the next metadata request.
+ // taking a request id so a local failure leaves the counter untouched.
namespace, err := NamespaceForRequest(code, payload, operation)
if err != nil {
return err
@@ -71,9 +70,10 @@ func StampRequestHeader(session *Session, code uint32, frame
[]byte) error {
default:
sessionID = session.SessionID()
if IsPartition(operation) {
- // Partition operations replicate in their own
per-partition group
- // with no client-table dedup, so they read the
watermark without
- // consuming it.
+ // Partition operations replicate in per-partition
groups that
+ // keep no client table, so there is nothing to
deduplicate
+ // against: the watermark is read without being
consumed and a
+ // partition-plane replay is at-least-once.
request = session.CurrentRequestID()
} else if request, err = session.NextRequestID(); err != nil {
return err
diff --git a/foreign/go/internal/vsr/header.go
b/foreign/go/internal/vsr/header.go
index 39d33df14..99cc742f9 100644
--- a/foreign/go/internal/vsr/header.go
+++ b/foreign/go/internal/vsr/header.go
@@ -49,6 +49,7 @@ const (
const (
replyOffsetSize = 48
replyOffsetCommand = 60
+ replyOffsetRequest = 200
replyOffsetOperation = 208
replyOffsetNamespace = 216
replyOffsetStatus = 224
@@ -170,6 +171,18 @@ func ReadReplyOperation(header *[HeaderSize]byte)
Operation {
return Operation(header[replyOffsetOperation])
}
+// ReadReplyRequestID reads the request id the reply echoes from the request
+// it answers, which is how the transport correlates a reply with the request
+// in flight.
+func ReadReplyRequestID(header *[HeaderSize]byte) uint64 {
+ return binary.LittleEndian.Uint64(header[replyOffsetRequest:])
+}
+
+// StampedRequestID reads the request id back out of a stamped request frame.
+func StampedRequestID(frame []byte) uint64 {
+ return binary.LittleEndian.Uint64(frame[requestOffsetRequest:])
+}
+
// Eviction holds the eviction-header fields the client acts on.
type Eviction struct {
Reason EvictionReason
diff --git a/foreign/go/internal/vsr/protocol_parity_test.go
b/foreign/go/internal/vsr/protocol_parity_test.go
index 1c0a32206..f995ce25b 100644
--- a/foreign/go/internal/vsr/protocol_parity_test.go
+++ b/foreign/go/internal/vsr/protocol_parity_test.go
@@ -29,6 +29,7 @@ import (
"strings"
"testing"
+ ierror "github.com/apache/iggy/foreign/go/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -47,6 +48,8 @@ var rustSources = map[string]string{
"operation": "core/binary_protocol/src/consensus/operation.rs",
"namespace": "core/binary_protocol/src/namespace.rs",
"cargo": "core/binary_protocol/Cargo.toml",
+ "eviction": "core/common/src/error/eviction.rs",
+ "sdk": "core/sdk/src/vsr.rs",
}
// goOperations names every discriminant the codec declares. It exists so the
@@ -165,8 +168,10 @@ var rustFieldLayout = map[string][2]int{
"EvictionReason": {1, 1},
}
-// loadRustSources reads the protocol crate files, skipping the test when the
-// checkout is not available.
+// loadRustSources reads the protocol crate files. One sentinel path answers
+// "is this a full checkout" and is the only reason to skip; every other read
+// is required, so a moved or renamed Rust source fails the suite instead of
+// silently turning every parity assertion into a skip.
func loadRustSources(t *testing.T) map[string]string {
t.Helper()
@@ -174,12 +179,16 @@ func loadRustSources(t *testing.T) map[string]string {
require.NoError(t, err)
root := filepath.Clean(filepath.Join(workDir, "..", "..", "..", ".."))
+ sentinel := filepath.Join(root, "core", "binary_protocol", "Cargo.toml")
+ if _, err := os.Stat(sentinel); err != nil {
+ t.Skipf("not a full checkout, %s is missing: %v", sentinel, err)
+ }
+
sources := make(map[string]string, len(rustSources))
for name, relative := range rustSources {
content, err := os.ReadFile(filepath.Join(root, relative))
- if err != nil {
- t.Skipf("protocol sources are not available at %s: %v",
root, err)
- }
+ require.NoError(t, err,
+ "protocol source %s moved or vanished; update
rustSources", relative)
sources[name] = string(content)
}
return sources
@@ -505,6 +514,98 @@ func rustMatchesAllowlist(t *testing.T, source, predicate
string) map[string]str
return names
}
+// rustEvictionErrorCodes maps the IggyError variant names the canonical
+// eviction mapping uses to their wire codes.
+var rustEvictionErrorCodes = map[string]ierror.Code{
+ "InvalidCredentials": ierror.InvalidCredentialsCode,
+ "InvalidPersonalAccessToken": ierror.InvalidPersonalAccessTokenCode,
+ "Unauthenticated": ierror.UnauthenticatedCode,
+ "StaleClient": ierror.StaleClientCode,
+ "InvalidFormat": ierror.InvalidFormatCode,
+ "InvalidCommand": ierror.InvalidCommandCode,
+}
+
+func TestProtocolParity_EvictionReasonMapping(t *testing.T) {
+ sources := loadRustSources(t)
+ body := captureBlock(sources["eviction"], `match reason \{`)
+ require.NotEmpty(t, body, "the eviction_reason_to_error match was not
found")
+
+ armPattern := regexp.MustCompile(
+
`(?s)((?:EvictionReason::[A-Za-z0-9]+\s*\|?\s*)+)=>\s*IggyError::([A-Za-z0-9]+)`)
+ reasonPattern := regexp.MustCompile(`EvictionReason::([A-Za-z0-9]+)`)
+
+ checked := 0
+ for _, arm := range armPattern.FindAllStringSubmatch(body, -1) {
+ wantCode, ok := rustEvictionErrorCodes[arm[2]]
+ require.True(t, ok, "IggyError::%s is not in the parity table",
arm[2])
+ for _, reason := range
reasonPattern.FindAllStringSubmatch(arm[1], -1) {
+ value, ok := goEvictionReasons[reason[1]]
+ require.True(t, ok, "unknown eviction reason %s",
reason[1])
+ mapped := NewEvictionError(Eviction{Reason: value})
+ assert.Equal(t, wantCode, mapped.Code(),
"EvictionReason::%s", reason[1])
+ checked++
+ }
+ }
+ require.GreaterOrEqual(t, checked, 8, "the mapping arms did not parse")
+
+ // The fallback arm has teeth: a reason byte from a newer server must
map
+ // the same way on both sides, and never to a reconnectable error.
+ fallback := regexp.MustCompile(`_ =>
IggyError::([A-Za-z0-9]+)`).FindStringSubmatch(body)
+ require.NotNil(t, fallback, "the fallback arm was not found")
+ wantFallback, ok := rustEvictionErrorCodes[fallback[1]]
+ require.True(t, ok, "IggyError::%s is not in the parity table",
fallback[1])
+ mapped := NewEvictionError(Eviction{Reason: EvictionReason(0xEE)})
+ assert.Equal(t, wantFallback, mapped.Code(),
+ "the unknown-reason fallback diverges from core/common")
+ // IncompatibleProtocol carries its own window logic, covered by the
+ // dedicated eviction tests in reply_test.go.
+}
+
+func TestProtocolParity_NamespaceRouting(t *testing.T) {
+ sources := loadRustSources(t)
+ codeValues := rustCommandCodes(sources["codes"])
+ require.NotEmpty(t, codeValues)
+
+ body := captureBlock(sources["sdk"], `fn namespace_for_request\(`)
+ require.NotEmpty(t, body, "the Rust namespace_for_request routing was
not found")
+
+ armed := make(map[uint32]string)
+ for _, arm := range regexp.MustCompile(`(?m)^\s*([A-Z0-9_]+_CODE) =>
\{`).
+ FindAllStringSubmatch(body, -1) {
+ value, ok := codeValues[arm[1]]
+ require.True(t, ok, "unknown command constant %s", arm[1])
+ armed[uint32(value)] = arm[1]
+ }
+ require.NotEmpty(t, armed, "no payload-peek arms were parsed")
+
+ for name, value := range codeValues {
+ code := uint32(value)
+ operation := OperationForCode(code)
+ if operation == OperationRegister || operation ==
OperationLogout ||
+ operation == OperationNonReplicated ||
IsMetadata(operation) {
+ continue
+ }
+ // A code that reaches the payload peek fails on an empty
payload; a
+ // code Rust does not route is refused outright. The two errors
keep
+ // the routing decisions distinguishable without crafting
payloads.
+ _, err := NamespaceForRequest(code, nil, operation)
+ if _, peeked := armed[code]; peeked {
+ assert.ErrorIs(t, err, ierror.ErrInvalidCommand,
+ "%s must derive its namespace from the
payload", name)
+ } else {
+ assert.ErrorIs(t, err, ierror.ErrFeatureUnavailable,
+ "%s must be refused rather than routed
blindly", name)
+ }
+ }
+
+ for code, name := range armed {
+ operation := OperationForCode(code)
+ shortCircuited := operation == OperationRegister || operation
== OperationLogout ||
+ operation == OperationNonReplicated ||
IsMetadata(operation)
+ assert.False(t, shortCircuited, "%s never reaches the namespace
peek in Go", name)
+ }
+}
+
func TestProtocolParity_PackedProtocolVersion(t *testing.T) {
sources := loadRustSources(t)
pattern := regexp.MustCompile(`(?m)^version =
"([0-9]+)\.([0-9]+)\.([0-9]+)`)
diff --git a/foreign/go/internal/vsr/reply.go b/foreign/go/internal/vsr/reply.go
index 16b72b2af..0c6d8cdc5 100644
--- a/foreign/go/internal/vsr/reply.go
+++ b/foreign/go/internal/vsr/reply.go
@@ -157,12 +157,12 @@ func NewEvictionError(eviction Eviction) *EvictionError {
}
case EvictionMalformedLogin:
mapped = ierror.ErrInvalidFormat
- case EvictionReserved, EvictionClientReleaseTooLow,
EvictionClientReleaseTooHigh,
- EvictionInvalidRequestOperation, EvictionInvalidRequestBody,
- EvictionInvalidRequestBodySize:
- mapped = ierror.ErrInvalidCommand
default:
- mapped = ierror.ErrUnauthenticated
+ // Everything else, including a reason byte this SDK does not
know,
+ // maps to InvalidCommand like the core/common fallback. The
unknown
+ // byte must not map to a reconnectable error, or a newer
server's new
+ // reason silently drives a disconnect and re-login loop.
+ mapped = ierror.ErrInvalidCommand
}
return &EvictionError{
Reason: eviction.Reason,
diff --git a/foreign/go/internal/vsr/reply_test.go
b/foreign/go/internal/vsr/reply_test.go
index 7fe4d5fef..a6755613f 100644
--- a/foreign/go/internal/vsr/reply_test.go
+++ b/foreign/go/internal/vsr/reply_test.go
@@ -231,7 +231,9 @@ func TestNewEvictionError_MapsEveryReason(t *testing.T) {
{reason: EvictionSessionError, want:
ierror.UnauthenticatedCode},
{reason: EvictionStaleClient, want: ierror.StaleClientCode},
{reason: EvictionMalformedLogin, want:
ierror.InvalidFormatCode},
- {reason: EvictionReason(200), want: ierror.UnauthenticatedCode},
+ // An unrecognized reason maps like the core/common fallback
and must
+ // not become a reconnectable error.
+ {reason: EvictionReason(200), want: ierror.InvalidCommandCode},
}
for _, test := range tests {
got := NewEvictionError(Eviction{Reason: test.reason})
diff --git a/foreign/go/internal/vsr/session.go
b/foreign/go/internal/vsr/session.go
index e44a3ac03..78226f4a3 100644
--- a/foreign/go/internal/vsr/session.go
+++ b/foreign/go/internal/vsr/session.go
@@ -128,9 +128,14 @@ func (s *Session) NextRequestID() (uint64, error) {
}
// CurrentRequestID returns the watermark without advancing it. Non-replicated
-// and partition-plane requests use it: the server's client table tracks ids
-// only for replicated metadata, so consuming one here would gap the next
-// metadata request and the primary would drop it.
+// and partition-plane requests use it because neither consults the client
+// table: non-replicated requests route by transport identity, and the
+// partition plane replicates in per-partition groups with no dedup table at
+// all. The table accepts any id above the watermark with no contiguity
+// requirement, so consuming one here would not gap anything; the invariant
+// that matters is that every partition request on a session carries the id
+// the next metadata operation will claim, and that a partition-plane replay
+// is therefore at-least-once.
func (s *Session) CurrentRequestID() uint64 {
return s.requestCounter
}
