gRPC 채팅 스트림에서 트랜잭션이 적용되지 않았던 이유(feat. 자기 호출, 프록시)

🧩문제 상황

 

 

 

 

 

 

 

채팅방 목록을 조회할 때, 각 채팅방에 대해 가장 최근 메시지(latestMessage)와 그 시간(lastestTime)을 함께 보여줘야 했습니다.

하지만 실제 화면에서는 최근 메시지가 표시되지 않는 문제가 발생했습니다.

 

 

 

 

 

 

🔍 초기 설계

채팅 메시지를 보낼 때, 해당 채팅방의 최근 메시지 정보를 함께 업데이트하도록 설계했습니다.

이를 위해 채팅 처리 메서드에 @Transactional을 붙이고, 내부에서 updateLatest()라는 메서드를 호출해 채팅방 정보를 갱신하도록했습니다.

 

⚠️ 이상 현상

하지만 로그를 확인해보니 메시지 저장(INSERT)은 정상적으로 수행되었지만,
채팅방의 최근 메시지 업데이트(UPDATE)는 실행되지 않았습니다.

아래 쿼리를 보면 메시지를 insert하고 채팅방 업데이트 쿼리는 실행하지 않은 것을 알 수 있습니다.

[Hibernate] 
    select
        c1_0.id,
        c1_0.created_at,
        c1_0.deleted_at,
        c1_0.latest_content,
        c1_0.latest_content_time,
        c1_0.product_id,
        c1_0.receiver_id,
        c1_0.room_name,
        c1_0.sender_id,
        c1_0.updated_at 
    from
        chatroom c1_0 
    where
        c1_0.id=?
[Hibernate] 
    insert 
    into
        message
        (chatroom_id, content, created_at, deleted_at, receiver_id, sender_id, updated_at) 
    values
        (?, ?, ?, ?, ?, ?, ?)

 

❓ 원인 분석

 

여기서 핵심 이슈는 바로 Spring의 @Transactional 자기호출(self-invocation) 문제였습니다.
채팅 메시지를 처리하는 gRPC 스트림의 onNext() 메서드 내부에서 같은 클래스의 updateLatest()를 호출했기 때문에, Spring AOP 프록시를 경유하지 않고 원본 인스턴스를 직접 호출하게 되었고, 결과적으로 @Transactional이 적용되지 않아 트랜잭션이 시작되지 않았습니다.

 

 

🚩에러가 나는 코드

public class ChatServiceImpl implements ChatService  {

    private final ChatroomRepository chatroomRepository;
    private final ChatRepository chatRepository;
    private final ProductRepository productRepository;
    private final ProfileGrpcClient profileGrpcClient;
 
    //채팅 메시지 전송
    @Transactional
    @Override
    public StreamObserver<Chat.ChatMessage> chat(StreamObserver<Chat.ChatResponseMessage> responseObserver) {
        return new StreamObserver<>() {
            private Long currentUserId;

            @Override
            public void onNext(Chat.ChatMessage msg) {
                log.info("product-service gRPC 수신: {}", msg);
                currentUserId = msg.getSenderId();
                userStreamMap.put(currentUserId, responseObserver);
                // grpc -> dto
                ChatMessageRequestDTO dto = ChatMessageRequestDTO.from(msg);

                Message message = Message.from(dto); //엔티티로 바꿈
                //이때 채팅방 업데이트 -> 여기서 트랜잭셔널 자기호출 문제가 생김.
                updateLatest(message.getChatroomId(), message.getContent(), message.getCreatedAt());
                
                Message saved = chatRepository.save(message);

                // 공통 응답 생성
                Chat.ChatResponseMessage response = Chat.ChatResponseMessage.newBuilder()
                        .setChatroomId(msg.getChatroomId())
                        .setSenderId(msg.getSenderId())
                        .setReceiverId(msg.getReceiverId())
                        .setContent(msg.getContent())
                        .setTimestamp(System.currentTimeMillis())
                        .setMessageId(saved.getId()) // DB 생성된 메시지 ID
                        .build();
               
               // 발신자 본인 세션에도 메시지 전달
                responseObserver.onNext(response);

                StreamObserver<Chat.ChatResponseMessage> receiverObserver = userStreamMap.get(msg.getReceiverId());
                if (receiverObserver != null) {
                    receiverObserver.onNext(response); // ← Gateway로 메시지 전달
                }
            }

            @Override
            public void onError(Throwable t) {
                userStreamMap.remove(currentUserId);
            }

            @Override
            public void onCompleted() {
                userStreamMap.remove(currentUserId);
                responseObserver.onCompleted();
            }
        };
    } 
    
    //최근 메시지 업데이트
    @Transactional
    @Override
    public void updateLatest(Long chatroomId, String content, LocalDateTime time) {
        Chatroom room = chatroomRepository.findById(chatroomId)
                .orElseThrow(()->new CustomException(ErrorCode.CHATROOM_NOT_FOUND));
        room.updateLatest(content,time);
    }
}

 

🔁 코드 흐름 설명

public StreamObserver<Chat.ChatMessage> chat(StreamObserver<Chat.ChatResponseMessage> responseObserver)

• 클라이언트가 gRPC로 채팅을 시작하면, 서버의 chat() 메서드가 호출됩니다.
• 이 chat() 메서드는 클라이언트가 보낼 메시지를 받을 준비가 된 객체(StreamObserver) 를 만들어서 리턴합니다.
• 클라이언트가 메시지를 보낼 때마다, 그 객체의 onNext() 메서드가 호출됩니다.

            @Override
            public void onNext(Chat.ChatMessage msg) {
                log.info("product-service gRPC 수신: {}", msg);
                currentUserId = msg.getSenderId();
                userStreamMap.put(currentUserId, responseObserver);
                // grpc -> dto
                ChatMessageRequestDTO dto = ChatMessageRequestDTO.from(msg);

                Message message = Message.from(dto); //엔티티로 바꿈
                //이때 채팅방 업데이트 -> 여기서 트랜잭셔널 자기호출 문제가 생김.
                updateLatest(message.getChatroomId(), message.getContent(), message.getCreatedAt());
                
                Message saved = chatRepository.save(message);

                // 공통 응답 생성
                Chat.ChatResponseMessage response = Chat.ChatResponseMessage.newBuilder()
                        .setChatroomId(msg.getChatroomId())
                        .setSenderId(msg.getSenderId())
                        .setReceiverId(msg.getReceiverId())
                        .setContent(msg.getContent())
                        .setTimestamp(System.currentTimeMillis())
                        .setMessageId(saved.getId()) // DB 생성된 메시지 ID
                        .build();
               
               // 발신자 본인 세션에도 메시지 전달
                responseObserver.onNext(response);

                StreamObserver<Chat.ChatResponseMessage> receiverObserver = userStreamMap.get(msg.getReceiverId());
                if (receiverObserver != null) {
                    receiverObserver.onNext(response); // ← Gateway로 메시지 전달
                }
            }

 

• onNext()를 Override하여 원하는 동작을 하도록 구현해줬습니다.

🔧 그래서 이 코드의 의미는?

내부에서 StreamObserver를 리턴 → 이후 클라이언트가 메시지를 보낼 때마다 onNext()가 호출

 

서버는 클라이언트의 메시지를 받을 수 있는 리스너(StreamObserver)를 만들어서 넘겨주고,
클라이언트가 메시지를 보낼 때마다 그 리스너의 onNext()가 실행되는 것입니다.

 

🔍 기존 코드에서의 트랜잭션 흐름

1. chat() 메서드에 붙은 @Transactional은 효과 없습니다.

• chat()은 gRPC 스트림을 리턴하는 메서드이기 때문입니다. -> 스트림을 리턴하고 트랜잭션이 종료됩니다.
• 실제 메시지를 처리하는 로직은 onNext()에서 실행됩니다.
• 즉, chat()에 붙은 트랜잭션은 이미 닫힌 상태에서 onNext()가 실행됩니다.→ 트랜잭션 없음.

2. onNext()는 gRPC 런타임이 다른 스레드에서 비동기적으로 호출합니다.

• chat()의 트랜잭션 범위 밖에서 실행됩니다.
• 즉, onNext() 자체는 Spring 트랜잭션이 없는 상태에서 실행됩니다.

2. onNext() 내부에서 같은 클래스의 updateLatest()을 호출합니다.

• Spring의 @Transactional은 프록시 객체를 통해 트랜잭션을 적용합니다.
• 그런데 같은 클래스 안에서 this.updateLatest()처럼 호출하면, 프록시를 거치지 않고 자기 자신을 직접 호출하게 됩니다.
• 결과적으로 @Transactional이 무시되고→ 트랜잭션이 시작되지 않습니다.

💥 그래서 어떤 일이 벌어지나?

• chatRepository.save(message)는 Spring Data JPA의 내부 트랜잭션으로 INSERT는 됩니다.
• 하지만 updateLatest()는 트랜잭션 없이 실행되니까, 엔티티 필드를 바꿔도 Dirty Checking이 작동하지 않아서 UPDATE 쿼리가 나가지 않았던 것입니다.
  • Dirty Checking : JPA가 트랜잭션 안에서 엔티티의 변경을 자동으로 감지해서 UPDATE쿼리를 날리는 기능

🛠️ 해결 방법

• updateLatest()를 다른 @Service 클래스로 분리해서,
  onNext()에서 프록시를 경유하도록 호출하면 @Transactional이 정상 작동합니다.
• 아래 처럼 구현하여 INSERT + UPDATE를 하나의 트랜잭션으로 묶어서 처리했습니다.

@Service
@RequiredArgsConstructor
public class ChatMessageTxServiceImpl implements ChatMessageTxService {

    public final ChatroomRepository chatroomRepository;
    public final ChatRepository chatRepository;

    @Transactional
    @Override
    public Message handleMessage(ChatMessageRequestDTO dto) {
        Message message = Message.from(dto); //엔티티로 바꿈
        Message saved = chatRepository.save(message); // 여기서 createdAt 세팅

        Chatroom room = chatroomRepository.findById(dto.getChatroomId())
                .orElseThrow(() -> new CustomException(ErrorCode.CHATROOM_NOT_FOUND));
        room.updateLatest(message.getContent(), saved.getCreatedAt() != null ? saved.getCreatedAt() : LocalDateTime.now());

        return saved;
    }
}

 

onNext()

Message saved = txService.handle(dto); // 프록시 경유 → 트랜잭션 보장

 

 

 

 

 

 

 

 

처리 후 다시 포스트맨에서 조회를 해보니,

최신 메시지 조회가 잘 되는 것을 확인할 수 있었습니다.

 

 

 

 

 

 

 

 

 

 

 

 

 

📍문제 핵심 요약

 

1. onNext()는 스프링 빈이 아니기 때문에 프록시가 적용되지 않습니다.

    @Transactional은 스프링이 만든 프록시를 통해 동작합니다.

    하지만 onNext()는 gRPC 런타임이 만든 익명 내부 클래스 인스턴스의 메서드이기 때문에 스프링 컨테이너가 관리하지 않습니다.

 

2. 자기호출로 프록시 우회

    기존에 onNext()안에서 같은 클래스의 updateLatest()를 호출했고,

    updateLatest()에 @Transactional을 적용했습니다.

    동일 클래스 내부 호출은 프록시를 거치지 않고 원본 메서드를 직접 호출합니다.

    즉, @Transactional이 무시되어서 트랜잭션이 시작되지 않습니다.

 

3. Dirty Checking 실패

    chatRepository.save(message)는 리포지토리 메서드 자체가 @Transactional(readOnly=false)이라 한번의 호출동안만 내부     트랜잭션이 열려 Insert가 됩니다.

    하지만 그 바깥에서 room.updateLatest()를 하면, 해당 시점에 활성 트랜잭션/영속성 컨텍스트가 없으므로

JPA Dirty Checking이 작동하지 않아서 Update가 나가지 않았던 것입니다.

 

 

채팅메시지 저장은 Spring 내부 트랜잭션이 열려서 가능했던거고 update는 트랜잭션이 열려야하는데 같은 클래스 내에서 메서드를 호출해서 자기호출문제로 트랜잭션이 열리지 않았던거고 더티 체크를 하지 못해서 업데이트가 안된겁니다.

 

✔️ 해결 요약

@Service로 분리하고, 외부 클래스에서 이 Bean의 메서드를 호출하면
Spring이 내부적으로 프록시 객체를 통해 트랜잭션을 자동으로 열고 닫아줍니다.

 

 

🧭 느낀점

이번 문제를 통해 단순히 "@Transactional만 붙이면 되는게 아니다"라는 것을 몸소 느꼈습니다.

Spring의 AOP 프록시 구조, 트랜잭션 경계, JPA의 Dirty Checking이 서로 맞물려 돌아가야 데이터의 일관성이 보장된다는 것을 알게 되었습니다.