refactor(go): seperate transport state and session state.

Author: chengxilo

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,12 +433,19 @@ 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() {
 	if c.conn != nil {
 		_ = c.conn.Close()
 	}
-	c.state = iggcon.StateDisconnected
+	c.transportState = iggcon.TransportStateDisconnected
+	c.sessionState = iggcon.SessionStateUnauthenticated
 }
 
 func (c *IggyTcpClient) GetConnectionInfo() *iggcon.ConnectionInfo {
@@ -451,24 +460,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()
@@ -543,7 +550,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.")
@@ -554,7 +561,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()
@@ -610,12 +617,13 @@ func (c *IggyTcpClient) disconnect() error {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 
-	if c.state == iggcon.StateDisconnected {
+	if c.transportState == iggcon.TransportStateDisconnected {
 		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
 
 	if c.conn != nil {
 		if err := c.conn.Close(); err != nil {
@@ -632,7 +640,7 @@ func (c *IggyTcpClient) shutdown() error {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 
-	if c.state == iggcon.StateShutdown {
+	if c.transportState == iggcon.TransportStateShutdown {
 		return nil
 	}
 
@@ -644,7 +652,8 @@ func (c *IggyTcpClient) shutdown() error {
 		}
 	}
 
-	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 nil

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)
 	}
 }
 
@@ -272,3 +273,23 @@ func TestNewIggyTcpClient_StoresProvidedLogger(t *testing.T) {
 		t.Errorf("expected logger output to contain 'source=tcp', got: %q", output)
 	}
 }
+
+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)
+	}
+}

foreign/go/client/tcp/tcp_session_management.go

@@ -30,41 +30,42 @@ 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))
+	c.mtx.Lock()
+	pre := c.sessionState
+	c.sessionState = iggcon.SessionStateAuthenticating
+	c.mtx.Unlock()
+
+	buffer, err := c.do(ctx, loginCmd)
 	if err != nil {
+		// A rejected login leaves the session state untouched, so restore the
+		// pre-login state only while it is still Authenticating: if the state
+		// moved, the connection may died mid-attempt and invalidation already
+		// recorded it unauthenticated, skip the restoration.
+		c.mtx.Lock()
+		if c.sessionState == iggcon.SessionStateAuthenticating {
+			c.sessionState = pre
+		}
+		c.mtx.Unlock()
 		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 +74,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) {

foreign/go/contracts/state.go

@@ -17,30 +17,45 @@
 
 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:
+	default:
+		return "unknown"
+	}
+}
+
+type SessionState uint8
+
+const (
+	SessionStateUnauthenticated SessionState = iota
+	SessionStateAuthenticating
+	SessionStateAuthenticated
+)
+
+func (s SessionState) String() string {
+	switch s {
+	case SessionStateUnauthenticated:
+		return "unauthenticated"
+	case SessionStateAuthenticating:
 		return "authenticating"
-	case StateAuthenticated:
+	case SessionStateAuthenticated:
 		return "authenticated"
 	default:
 		return "unknown"