Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions gradle/libs.versions.toml
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ kotlin = "2.4.20"
kotest = "6.2.5"
ktlint = "12.1.2"
exposed = "1.5.0"
jackson = "2.22.2"
jackson = "2.22.3"
liquibase = "5.0.4"
testcontainers = "1.21.4"
postgresql = "42.7.13"
Expand All @@ -13,10 +13,10 @@ spring = "7.0.9"
springBoot = "4.1.1"
vanniktechPublish = "0.37.0"
wiremock = "3.13.2"
slf4j = "2.0.19"
logback = "1.6.3"
slf4j = "2.0.20"
logback = "1.6.4"
assertj = "3.27.7"
h2 = "2.5.250"
h2 = "2.5.252"
micrometer = "1.17.1"
jmh = "1.37"
jmhGradlePlugin = "0.7.3"
Expand All @@ -25,7 +25,7 @@ pitestTool = "1.17.0"
pitestJunit5Plugin = "1.2.1"
# Hibernate ORM 7.x (the line Spring Framework 7.x targets); integration-tests only, to exercise
# JpaTransactionManager fail-fast extraction.
hibernate = "7.4.10.Final"
hibernate = "7.4.11.Final"

[libraries]
kotlinGradlePlugin = { module = "org.jetbrains.kotlin:kotlin-gradle-plugin", version.ref = "kotlin" }
Expand Down
Binary file modified gradle/wrapper/gradle-wrapper.jar
Binary file not shown.
2 changes: 1 addition & 1 deletion gradle/wrapper/gradle-wrapper.properties
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
distributionUrl=https\://services.gradle.org/distributions/gradle-9.7.1-bin.zip
distributionUrl=https\://services.gradle.org/distributions/gradle-9.8.0-bin.zip
networkTimeout=10000
retries=0
retryBackOffMs=500
Expand Down
52 changes: 41 additions & 11 deletions gradlew.bat

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Original file line number Diff line number Diff line change
@@ -1,19 +1,21 @@
package com.softwaremill.okapi.http

import com.github.tomakehurst.wiremock.WireMockServer
import com.github.tomakehurst.wiremock.client.WireMock.aResponse
import com.github.tomakehurst.wiremock.client.WireMock.post
import com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo
import com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig
import com.softwaremill.okapi.core.DeliveryResult
import com.softwaremill.okapi.core.OutboxEntry
import com.softwaremill.okapi.core.OutboxMessage
import com.sun.net.httpserver.HttpServer
import io.kotest.assertions.withClue
import io.kotest.core.spec.style.FunSpec
import io.kotest.matchers.longs.shouldBeLessThan
import io.kotest.matchers.shouldBe
import java.net.InetSocketAddress
import java.time.Instant
import java.util.concurrent.BrokenBarrierException
import java.util.concurrent.CyclicBarrier
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException

private const val SCHEDULING_JITTER_TOLERANCE_MS = 200L
private const val RENDEZVOUS_TIMEOUT_SECONDS = 10L

private fun entry(suffix: String): OutboxEntry {
val info = httpDeliveryInfo {
Expand All @@ -23,46 +25,46 @@ private fun entry(suffix: String): OutboxEntry {
return OutboxEntry.createPending(OutboxMessage("evt-$suffix", """{"k":"v-$suffix"}"""), info, Instant.now())
}

/**
* Proves `deliverBatch` fires requests concurrently rather than one-at-a-time, using WireMock's
* request journal (`allServeEvents`) rather than just overall wall-clock time — the timestamp
* spread across requests is direct evidence they overlapped in flight.
*/
class HttpMessageDelivererBatchConcurrencyTest : FunSpec({
val wiremock = WireMockServer(wireMockConfig().dynamicPort())
val deliverer by lazy {
HttpMessageDeliverer({ "http://localhost:${wiremock.port()}" })
}

beforeSpec { wiremock.start() }
afterSpec { wiremock.stop() }
beforeEach { wiremock.resetAll() }

test("deliverBatch fires N requests concurrently: request-log timestamps overlap within one delay window") {
test("deliverBatch keeps every request in flight before any response completes") {
val batchSize = 10
val delayMs = 300L
wiremock.stubFor(post(urlEqualTo("/test")).willReturn(aResponse().withStatus(200).withFixedDelay(delayMs.toInt())))
val entries = (1..batchSize).map { entry("e$it") }

val start = System.nanoTime()
val results = deliverer.deliverBatch(entries)
val elapsedMs = (System.nanoTime() - start) / 1_000_000
val barrier = CyclicBarrier(batchSize)
Executors.newVirtualThreadPerTaskExecutor().use { executor ->
val server = HttpServer.create(InetSocketAddress("127.0.0.1", 0), 0)
server.executor = executor
server.createContext("/test") { exchange ->
exchange.use {
it.requestBody.use { body -> body.readAllBytes() }
// Every handler waits for the whole batch before responding. Sequential
// delivery breaks the barrier and gets 503s; parallel delivery gets 200s.
val status = try {
barrier.await(RENDEZVOUS_TIMEOUT_SECONDS, TimeUnit.SECONDS)
200
} catch (_: InterruptedException) {
Thread.currentThread().interrupt()
503
} catch (_: BrokenBarrierException) {
503
} catch (_: TimeoutException) {
503
}
it.sendResponseHeaders(status, -1)
}
}

results.forEach { (_, r) -> r shouldBe DeliveryResult.Success }
try {
server.start()
val deliverer = HttpMessageDeliverer({ "http://127.0.0.1:${server.address.port}" })
val entries = (1..batchSize).map { entry("e$it") }

// Sequential delivery would take ~batchSize * delayMs; parallel delivery should land near
// one delay window plus scheduling overhead — well under half the sequential bound.
elapsedMs.shouldBeLessThan(batchSize * delayMs / 2)
val results = deliverer.deliverBatch(entries)

// Direct evidence of overlap: every request's logged arrival time falls within a single
// delay window of each other, i.e. WireMock received all of them before the first one
// could possibly have completed and freed up a sequential caller to send the next.
val loggedTimestamps = wiremock.allServeEvents.map { it.request.loggedDate.time }
loggedTimestamps.size shouldBe batchSize
val spreadMs = loggedTimestamps.maxOrNull()!! - loggedTimestamps.minOrNull()!!
// A fully sequential implementation would spread these across ~(batchSize - 1) * delayMs
// (2700 ms here); a small tolerance above one delay window still clearly distinguishes
// "overlapped" from "sequential" while absorbing CI scheduling/connection-setup jitter.
spreadMs.shouldBeLessThan(delayMs + SCHEDULING_JITTER_TOLERANCE_MS)
withClue("All requests must reach the server before it sends any successful response") {
results.map { it.result } shouldBe List(batchSize) { DeliveryResult.Success }
}
} finally {
server.stop(0)
}
}
}
})
2 changes: 1 addition & 1 deletion okapi-integration-tests/build.gradle.kts
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ dependencies {
testImplementation(libs.kafkaClients)

// SLF4J for Testcontainers logging
testRuntimeOnly("org.slf4j:slf4j-simple:2.0.19")
testRuntimeOnly("org.slf4j:slf4j-simple:2.0.20")

// WireMock (HTTP E2E tests)
testImplementation(libs.wiremock)
Expand Down
Loading