This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 6ca219f67 refactor(go): separate transport state and session state
(#3652)
6ca219f67 is described below
commit 6ca219f6795769780a671e833939e54fb09ae15a
Author: Chengxi Luo <[email protected]>
AuthorDate: Fri Jul 17 04:48:17 2026 -0400
refactor(go): separate transport state and session state (#3652)
---
foreign/go/client/tcp/tcp_core.go | 50 ++++++++------
foreign/go/client/tcp/tcp_core_test.go | 88 ++++++++++++++++++++-----
foreign/go/client/tcp/tcp_session_management.go | 42 +++++-------
foreign/go/contracts/state.go | 42 +++++++-----
4 files changed, 146 insertions(+), 76 deletions(-)
diff --git a/foreign/go/client/tcp/tcp_core.go
b/foreign/go/client/tcp/tcp_core.go
index 2988db7db..8952fce1a 100644
--- a/foreign/go/client/tcp/tcp_core.go
+++ b/foreign/go/client/tcp/tcp_core.go
@@ -59,7 +59,8 @@ type IggyTcpClient struct {
clientAddress string
currentServerAddress string
connectedAt time.Time
- state iggcon.State
+ transportState iggcon.TransportState
+ sessionState iggcon.SessionState
// respHeader is the reused response-status read buffer; guarded by
c.mtx.
respHeader [ResponseInitialBytesLength]byte
}
@@ -216,7 +217,8 @@ func NewIggyTcpClient(logger *slog.Logger, options
...Option) *IggyTcpClient {
logger: logger,
clientAddress: "",
conn: nil,
- state: iggcon.StateDisconnected,
+ transportState: iggcon.TransportStateDisconnected,
+ sessionState: iggcon.SessionStateUnauthenticated,
connectedAt: time.Time{},
leaderRedirectionState: iggcon.LeaderRedirectionState{},
currentServerAddress: opts.config.serverAddress,
@@ -341,14 +343,14 @@ func (c *IggyTcpClient) sendWireAndFetchResponse(ctx
context.Context, wirePayloa
c.mtx.Lock()
defer c.mtx.Unlock()
- switch c.state {
- case iggcon.StateShutdown:
+ switch c.transportState {
+ case iggcon.TransportStateShutdown:
c.logger.Debug("Cannot send data. Client is shutdown.")
return nil, ierror.ErrClientShutdown
- case iggcon.StateDisconnected:
+ case iggcon.TransportStateDisconnected:
c.logger.Debug("Cannot send data. Client is not connected.")
return nil, ierror.ErrNotConnected
- case iggcon.StateConnecting:
+ case iggcon.TransportStateConnecting:
c.logger.Debug("Cannot send data. Client is still connecting.")
return nil, ierror.ErrNotConnected
}
@@ -431,10 +433,17 @@ func (c *IggyTcpClient) sendLocked(wirePayload []byte)
([]byte, error) {
return buffer, nil
}
+func (c *IggyTcpClient) setSessionState(state iggcon.SessionState) {
+ c.mtx.Lock()
+ c.sessionState = state
+ c.mtx.Unlock()
+}
+
// invalidateConnLocked closes the connection and marks it as disconnected
func (c *IggyTcpClient) invalidateConnLocked() {
_ = c.closeConnLocked()
- c.state = iggcon.StateDisconnected
+ c.transportState = iggcon.TransportStateDisconnected
+ c.sessionState = iggcon.SessionStateUnauthenticated
}
// closeConnLocked closes and drops the current connection.
@@ -459,24 +468,22 @@ func (c *IggyTcpClient) GetConnectionInfo()
*iggcon.ConnectionInfo {
// Connect establishes the TCP connection to the server.
func (c *IggyTcpClient) Connect(ctx context.Context) error {
c.mtx.Lock()
- switch c.state {
- case iggcon.StateShutdown:
+ switch c.transportState {
+ case iggcon.TransportStateShutdown:
c.mtx.Unlock()
c.logger.Debug("Cannot connect. Client is shutdown.")
return ierror.ErrClientShutdown
- case iggcon.StateConnected,
- iggcon.StateAuthenticating,
- iggcon.StateAuthenticated:
+ case iggcon.TransportStateConnected:
clientAddress := c.clientAddress
c.mtx.Unlock()
c.logger.Debug("Client is already connected.",
slog.String("client_address", clientAddress))
return nil
- case iggcon.StateConnecting:
+ case iggcon.TransportStateConnecting:
c.mtx.Unlock()
c.logger.Debug("Client is already connecting.")
return nil
default:
- c.state = iggcon.StateConnecting
+ c.transportState = iggcon.TransportStateConnecting
}
connectedAt := c.connectedAt
c.mtx.Unlock()
@@ -551,7 +558,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
return nil
}); err != nil {
c.mtx.Lock()
- c.state = iggcon.StateDisconnected
+ c.transportState = iggcon.TransportStateDisconnected
c.mtx.Unlock()
if !c.config.reconnection.enabled {
c.logger.Warn("Automatic reconnection is disabled.")
@@ -562,7 +569,7 @@ func (c *IggyTcpClient) Connect(ctx context.Context) error {
c.mtx.Lock()
c.conn = conn
- c.state = iggcon.StateConnected
+ c.transportState = iggcon.TransportStateConnected
c.connectedAt = time.Now()
c.logger.Info("Iggy client has connected to the Iggy server",
slog.String("client_address", c.clientAddress), slog.String("server_address",
c.currentServerAddress))
c.mtx.Unlock()
@@ -618,12 +625,14 @@ func (c *IggyTcpClient) disconnect() error {
c.mtx.Lock()
defer c.mtx.Unlock()
- if c.state == iggcon.StateDisconnected || c.state ==
iggcon.StateShutdown {
+ if c.transportState == iggcon.TransportStateDisconnected ||
c.transportState == iggcon.TransportStateShutdown {
return nil
}
c.logger.Info("Iggy client is disconnecting from server...",
slog.String("client_address", c.clientAddress))
- c.state = iggcon.StateDisconnected
+ c.transportState = iggcon.TransportStateDisconnected
+ c.sessionState = iggcon.SessionStateUnauthenticated
+
err := c.closeConnLocked()
c.logger.Info("Iggy client has disconnected from server.",
slog.String("client_address", c.clientAddress))
@@ -635,15 +644,16 @@ func (c *IggyTcpClient) shutdown() error {
c.mtx.Lock()
defer c.mtx.Unlock()
- if c.state == iggcon.StateShutdown {
+ if c.transportState == iggcon.TransportStateShutdown {
return nil
}
c.logger.Info("Shutting down the Iggy TCP client...",
slog.String("client_address", c.clientAddress))
err := c.closeConnLocked()
- c.state = iggcon.StateShutdown
+ c.transportState = iggcon.TransportStateShutdown
+ c.sessionState = iggcon.SessionStateUnauthenticated
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_test.go
b/foreign/go/client/tcp/tcp_core_test.go
index 4f87aeaff..f996f0298 100644
--- a/foreign/go/client/tcp/tcp_core_test.go
+++ b/foreign/go/client/tcp/tcp_core_test.go
@@ -42,9 +42,10 @@ func newTestClient(t *testing.T) (*IggyTcpClient, net.Conn) {
t.Helper()
serverConn, clientConn := net.Pipe()
c := &IggyTcpClient{
- conn: clientConn,
- state: iggcon.StateConnected,
- logger: slog.New(slog.DiscardHandler),
+ conn: clientConn,
+ transportState: iggcon.TransportStateConnected,
+ sessionState: iggcon.SessionStateUnauthenticated,
+ logger: slog.New(slog.DiscardHandler),
}
t.Cleanup(func() {
err := clientConn.Close()
@@ -130,8 +131,8 @@ func TestSendAndFetchResponse_DeadlineTimeout(t *testing.T)
{
t.Errorf("got %v, want context.DeadlineExceeded", err)
}
// After a timeout, the connection should be invalidated.
- if c.state != iggcon.StateDisconnected {
- t.Errorf("expected state %v, got %v", iggcon.StateDisconnected,
c.state)
+ if c.transportState != iggcon.TransportStateDisconnected {
+ t.Errorf("expected state %v, got %v",
iggcon.TransportStateDisconnected, c.transportState)
}
// TODO: revisit after reconnect implementation
@@ -158,8 +159,8 @@ func TestSendAndFetchResponse_CancelDuringIO(t *testing.T) {
t.Errorf("got %v, want context.Canceled", err)
}
// Connection should be invalidated after the I/O error.
- if c.state != iggcon.StateDisconnected {
- t.Errorf("expected state %v, got %v", iggcon.StateDisconnected,
c.state)
+ if c.transportState != iggcon.TransportStateDisconnected {
+ t.Errorf("expected state %v, got %v",
iggcon.TransportStateDisconnected, c.transportState)
}
}
@@ -214,8 +215,8 @@ func TestSendAndFetchResponse_ErrorStatus(t *testing.T) {
t.Errorf("got %v, want %v", err, ierror.ErrUnauthenticated)
}
// Connection should remain healthy after an application-level error.
- if c.state != iggcon.StateConnected {
- t.Errorf("expected state %v, got %v", iggcon.StateConnected,
c.state)
+ if c.transportState != iggcon.TransportStateConnected {
+ t.Errorf("expected state %v, got %v",
iggcon.TransportStateConnected, c.transportState)
}
}
@@ -231,8 +232,8 @@ func TestSendAndFetchResponse_SuccessEmptyBody(t
*testing.T) {
if len(result) != 0 {
t.Errorf("expected empty result, got %d bytes", len(result))
}
- if c.state != iggcon.StateConnected {
- t.Errorf("expected state %v, got %v", iggcon.StateConnected,
c.state)
+ if c.transportState != iggcon.TransportStateConnected {
+ t.Errorf("expected state %v, got %v",
iggcon.TransportStateConnected, c.transportState)
}
}
@@ -249,8 +250,8 @@ func TestSendAndFetchResponse_SuccessWithBody(t *testing.T)
{
if string(result) != string(body) {
t.Errorf("got %q, want %q", result, body)
}
- if c.state != iggcon.StateConnected {
- t.Errorf("expected state %v, got %v", iggcon.StateConnected,
c.state)
+ if c.transportState != iggcon.TransportStateConnected {
+ t.Errorf("expected state %v, got %v",
iggcon.TransportStateConnected, c.transportState)
}
}
@@ -273,6 +274,59 @@ func TestNewIggyTcpClient_StoresProvidedLogger(t
*testing.T) {
}
}
+func TestLoginUser_LoginAndLogout(t *testing.T) {
+ c, serverConn := newTestClient(t)
+
+ identity := make([]byte, 4)
+ binary.LittleEndian.PutUint32(identity, 42)
+
+ go func() {
+ serverRespond(t, serverConn, 0, identity)
+ // login always probes for a leader afterwards; it must be
answered or the call blocks.
+ serverRespond(t, serverConn,
uint32(ierror.FeatureUnavailableCode), nil)
+ serverRespond(t, serverConn, 0, nil)
+ }()
+
+ ctx := context.Background()
+ info, err := c.LoginUser(ctx, "iggy", "iggy")
+ if err != nil {
+ t.Fatalf("unexpected login error: %v", err)
+ }
+ if info.UserId != 42 {
+ t.Errorf("got user id %d, want 42", info.UserId)
+ }
+ if c.sessionState != iggcon.SessionStateAuthenticated {
+ t.Errorf("expected session %v after login, got %v",
iggcon.SessionStateAuthenticated, c.sessionState)
+ }
+
+ if err := c.LogoutUser(ctx); err != nil {
+ t.Fatalf("unexpected logout error: %v", err)
+ }
+ if c.sessionState != iggcon.SessionStateUnauthenticated {
+ t.Errorf("expected session %v after logout, got %v",
iggcon.SessionStateUnauthenticated, c.sessionState)
+ }
+}
+
+func TestLoginUser_RejectedReloginKeepsExistingSession(t *testing.T) {
+ c, serverConn := newTestClient(t)
+ c.sessionState = iggcon.SessionStateAuthenticated
+
+ go serverRespond(t, serverConn, uint32(ierror.InvalidCredentialsCode),
nil)
+
+ _, err := c.LoginUser(context.Background(), "other-user",
"wrong-password")
+ if err == nil {
+ t.Fatal("expected login error, got nil")
+ }
+ if !errors.Is(err, ierror.ErrInvalidCredentials) {
+ t.Errorf("got %v, want %v", err, ierror.ErrInvalidCredentials)
+ }
+ // The server rejects the login before touching the existing session,
+ // so the client must keep reporting the session it still has.
+ if c.sessionState != iggcon.SessionStateAuthenticated {
+ t.Errorf("expected session to stay authenticated after rejected
relogin, got %v", c.sessionState)
+ }
+}
+
var errCloseFailed = errors.New("close failed")
// failingCloseConn is a connection whose Close always fails, standing in for a
@@ -295,8 +349,8 @@ func TestShutdown_FailedCloseStillCompletesTeardown(t
*testing.T) {
if err := c.shutdown(); !errors.Is(err, errCloseFailed) {
t.Fatalf("got %v, want %v", err, errCloseFailed)
}
- if c.state != iggcon.StateShutdown {
- t.Errorf("expected state %v, got %v", iggcon.StateShutdown,
c.state)
+ if c.transportState != iggcon.TransportStateShutdown {
+ t.Errorf("expected state %v, got %v",
iggcon.TransportStateShutdown, c.transportState)
}
if c.conn != nil {
t.Error("expected the closed connection to be dropped")
@@ -320,8 +374,8 @@ func TestDisconnect_ShutdownClientIsNotResurrected(t
*testing.T) {
t.Fatalf("unexpected disconnect error: %v", err)
}
- if c.state != iggcon.StateShutdown {
- t.Errorf("expected state to stay %v, got %v",
iggcon.StateShutdown, c.state)
+ if c.transportState != iggcon.TransportStateShutdown {
+ t.Errorf("expected state to stay %v, got %v",
iggcon.TransportStateShutdown, c.transportState)
}
_, err := c.sendWireAndFetchResponse(context.Background(), emptyWireReq)
diff --git a/foreign/go/client/tcp/tcp_session_management.go
b/foreign/go/client/tcp/tcp_session_management.go
index 9aa65651a..12654de93 100644
--- a/foreign/go/client/tcp/tcp_session_management.go
+++ b/foreign/go/client/tcp/tcp_session_management.go
@@ -30,41 +30,32 @@ import (
)
func (c *IggyTcpClient) LoginUser(ctx context.Context, username string,
password string) (*iggcon.IdentityInfo, error) {
- c.logger.Info("Iggy client is signing in...",
slog.String("client_address", c.clientAddress))
- buffer, err := c.do(ctx, &command.LoginUser{
+ return c.login(ctx, &command.LoginUser{
Username: username,
Password: password,
})
- if err != nil {
- return nil, err
- }
-
- c.logger.Info("Iggy client has signed in successfully.",
slog.String("client_address", c.clientAddress))
- identity := binaryserialization.DeserializeLogInResponse(buffer)
- shouldRedirect, err := c.HandleLeaderRedirection(ctx)
- if err != nil {
- return nil, err
- }
- if shouldRedirect {
- if err = c.Connect(ctx); err != nil {
- return nil, err
- }
- return c.LoginUser(ctx, username, password)
- }
- return identity, nil
}
func (c *IggyTcpClient) LoginWithPersonalAccessToken(ctx context.Context,
token string) (*iggcon.IdentityInfo, error) {
- c.logger.Info("Iggy client is signing in...",
slog.String("client_address", c.clientAddress))
- buffer, err := c.do(ctx, &command.LoginWithPersonalAccessToken{
+ return c.login(ctx, &command.LoginWithPersonalAccessToken{
Token: token,
})
+}
+
+func (c *IggyTcpClient) login(ctx context.Context, loginCmd command.Command)
(*iggcon.IdentityInfo, error) {
+ c.logger.Info("Iggy client is signing in...",
slog.String("client_address", c.clientAddress))
+
+ // A failed login never writes the session state: a server-side reject
+ // leaves the existing session untouched, and a connection that dies
+ // mid-attempt is already reset to unauthenticated by
invalidateConnLocked.
+ buffer, err := c.do(ctx, loginCmd)
if err != nil {
return nil, err
}
c.logger.Info("Iggy client has signed in successfully.",
slog.String("client_address", c.clientAddress))
identity := binaryserialization.DeserializeLogInResponse(buffer)
+ c.setSessionState(iggcon.SessionStateAuthenticated)
shouldRedirect, err := c.HandleLeaderRedirection(ctx)
if err != nil {
return nil, err
@@ -73,14 +64,17 @@ func (c *IggyTcpClient) LoginWithPersonalAccessToken(ctx
context.Context, token
if err = c.Connect(ctx); err != nil {
return nil, err
}
- return c.LoginWithPersonalAccessToken(ctx, token)
+ return c.login(ctx, loginCmd)
}
return identity, nil
}
func (c *IggyTcpClient) LogoutUser(ctx context.Context) error {
- _, err := c.do(ctx, &command.LogoutUser{})
- return err
+ if _, err := c.do(ctx, &command.LogoutUser{}); err != nil {
+ return err
+ }
+ c.setSessionState(iggcon.SessionStateUnauthenticated)
+ return nil
}
func (c *IggyTcpClient) HandleLeaderRedirection(ctx context.Context) (bool,
error) {
diff --git a/foreign/go/contracts/state.go b/foreign/go/contracts/state.go
index d06f043b2..32966b6f1 100644
--- a/foreign/go/contracts/state.go
+++ b/foreign/go/contracts/state.go
@@ -17,30 +17,42 @@
package iggcon
-type State uint8
+type TransportState uint8
const (
- StateShutdown State = iota
- StateDisconnected
- StateConnecting
- StateConnected
- StateAuthenticating
- StateAuthenticated
+ TransportStateDisconnected TransportState = iota
+ TransportStateShutdown
+ TransportStateConnecting
+ TransportStateConnected
)
-func (s State) String() string {
+func (s TransportState) String() string {
switch s {
- case StateShutdown:
+ case TransportStateShutdown:
return "shutdown"
- case StateDisconnected:
+ case TransportStateDisconnected:
return "disconnected"
- case StateConnecting:
+ case TransportStateConnecting:
return "connecting"
- case StateConnected:
+ case TransportStateConnected:
return "connected"
- case StateAuthenticating:
- return "authenticating"
- case StateAuthenticated:
+ default:
+ return "unknown"
+ }
+}
+
+type SessionState uint8
+
+const (
+ SessionStateUnauthenticated SessionState = iota
+ SessionStateAuthenticated
+)
+
+func (s SessionState) String() string {
+ switch s {
+ case SessionStateUnauthenticated:
+ return "unauthenticated"
+ case SessionStateAuthenticated:
return "authenticated"
default:
return "unknown"
