NoSuchTransaction in reactive mongodb client when working with transactions

[fremarti]

Setup

Spring Boot 3.3.4 with org.springframework.boot:spring-boot-starter-webflux and org.springframework.boot:spring-boot-starter-data-mongodb-reactive. MongoDB version is 6.0.18.

MongoDB Config

Reactive mongo client is configured in configuration to activate transactional feature in mongo templates:

@Configuration
@EnableConfigurationProperties(MongoProperties::class)
class MongoConfiguration(
    private val mongoProperties: MongoProperties,
) : AbstractReactiveMongoConfiguration() {
    ...
    @Bean
    fun transactionManager(
        factory: ReactiveMongoDatabaseFactory?,
        properties: MongoProperties,
    ): ReactiveMongoTransactionManager {
        return ReactiveMongoTransactionManager(
            factory!!,
            TransactionOptions.builder().readPreference(ReadPreference.valueOf(properties.readPreference)).build(),
        )
    }
    ...

Docker Setup

For local and integration testing a mongodb is configured using docker compose. The db is configured as single node replica set. The here mentioned init script just runs rs.initiate(...) to register replica set. In the application properties the according connection string is set with mongodb://localhost:27017/?replicaSet=rs0.

services:
  mongo:
    image: mongo:6.0.18
    ports:
      - "27017:27017"
    volumes:
      - ./bin/mongodb-init-replica-set.sh:/docker-entrypoint-initdb.d/mongodb-init-replica-set.sh:ro
    command: ["mongod", "--replSet", "rs0", "--bind_ip_all"]
  ...

Application Code

I have an endpoint PUT /foo which should update multiple entries in a single collection. This update should be transactional. Before updating the entries, all entries are fetched by ids and some validation is done before updating the entries:

// FooController.kt
@RestController
class FooController(private val fooUseCase: FooUseCase) {
    ...
    @Transactional(label = ["mongo:readPreference=PRIMARY"])
    @PutMapping(
        value = ["/foo"],
        consumes = [MediaType.APPLICATION_JSON_VALUE],
    )
    fun foo(@RequestBody request: RequestDto): Mono<Void> {
        return fooUseCase
            .process(request)
            .doOnError { error ->
                logger.error("Failed", error)
            }
    }
}

// FooUseCase.kt
@Service
class FooUseCase(private val repo: FooRepository, private val factory: FooFactory) {
    fun process(request: RequestDto): Mono<Void> {
        return repo
            .findAllById(request.ids)
            .collectList()
            .flatMap { entries ->
                // Do some checks
                repo
                    .saveAll(entries.map { factory.from(request, it) }
                    .then()
            }
    }
}

Integration Tests

To test the transactional behavior I wrote a Spring Boot integration test. I leveraged coroutines to fire 100 requests concurrently against the endpoint using the web test client to make sure there are no side-effects.

@SpringBootTest(
    webEnvironment = RANDOM_PORT,
    properties = ["server.error.include-stacktrace=always"]
)
class FooIntegrationTest {

    @Autowired
    lateinit var webTestClient: WebTestClient

    @Autowired
    lateinit var fooUsecase: FooUseCase

    @Autowired
    lateinit var repo: FooRepository

    // Clean-up in @BeforeEach and @AfterEach

    @Test
    fun `should rollback`() = runTest {
        // 1. Store entries which should be updated in db
        // 2. Assert entries are there
        // 3. Run db command to set validator rule for specific id to enforce exception on db request without the need to mock something

        // 4. Run test using web client:
        val responseSpecs = (1..100).map {
            async {
                webTestClient
                    .put()
                    .uri {
                        it.path("/foo")
                    }
                    .body(Mono.just(request), RequestDto::class.java)
                    .header(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE)
                    .exchange()
                    .expectStatus().is5xxServerError
                    .expectBody()
                    .jsonPath("$.trace").value<String> { stackTrace ->
                        stackTrace.shouldContain("DataIntegrityViolationException")
                        stackTrace.shouldNotContain("NoSuchTransaction")
                    }

            }
        }
        responseSpecs.awaitAll()

        // 5. Validate original entries in db are not altered
    }

Unfortunately, I see side effects in the transactional behavior. On a random basis there is a MongoTransactionException thrown with NoSuchTransaction instead of the expected DataIntegrityViolationException. Therefore this test fails and I cannot explain why that is. Can anybody help?

[fremarti]

What I forgot to mention:
Originally, I had the transactional annotation attached to the service method. This is when I recognized the side effects, so I adjusted that.
Besides the option with the @Transactional annotation I also tried to use the TransactionalOperator and wrapped the according chains in transactionOperator.execute { ... }, but didn't see any difference.

[fremarti]

I either see that the transaction has already been aborted which imo means that requests are sharing a transaction:

org.springframework.data.mongodb.MongoTransactionException: Command failed with error 251 (NoSuchTransaction): 'Transaction with { txnNumber: 67 } has been aborted.' on server localhost:27017. The full response is {"errorLabels": ["TransientTransactionError"], "ok": 0.0, "errmsg": "Transaction with { txnNumber: 67 } has been aborted.", "code": 251, "codeName": "NoSuchTransaction", "$clusterTime": {"clusterTime": {"$timestamp": {"t": 1728556211, "i": 10}}, "signature": {"hash": {"$binary": {"base64": "AAAAAAAAAAAAAAAAAAAAAAAAAAA=", "subType": "00"}}, "keyId": 0}}, "operationTime": {"$timestamp": {"t": 1728556211, "i": 10}}}

Or I see NoSuchTransaction with the message that the txNumber is not matching which seams like it is not finding the correct transaction in the Reactor context.

Error disappears when I set max connection pool size in the mongo config to 1.