Skip to content

Commit fb737ce

Browse files
slinkydeveloperslinkydeveloper
committed
Idempotent requests all around
1 parent 5f10a39 commit fb737ce

23 files changed

+208
-142
lines changed

‎src/main/kotlin/dev/restate/sdktesting/tests/AwaitTimeout.kt‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,8 @@ class AwaitTimeout {
4444
TestUtilsServiceClient.fromClient(ingressClient)
4545
.createAwakeableAndAwaitIt(
4646
CreateAwakeableAndAwaitItRequest(
47-
UUID.randomUUID().toString(), timeout.toMillis())))
47+
UUID.randomUUID().toString(), timeout.toMillis()),
48+
idempotentCallOptions()))
4849
.isEqualTo(TimeoutResponse)
4950
}
5051
}

‎src/main/kotlin/dev/restate/sdktesting/tests/CallOrdering.kt‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -73,9 +73,10 @@ class CallOrdering {
7373
} else {
7474
ManyCallRequest(proxyRequest, false, true)
7575
}
76-
})
76+
},
77+
idempotentCallOptions())
7778

78-
assertThat(ListObjectClient.fromClient(ingressClient, listName).clear())
79+
assertThat(ListObjectClient.fromClient(ingressClient, listName).clear(idempotentCallOptions()))
7980
.containsExactly("0", "1", "2")
8081
}
8182
}

‎src/main/kotlin/dev/restate/sdktesting/tests/CancelInvocation.kt‎

Lines changed: 21 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -19,8 +19,9 @@ import java.util.*
1919
import kotlin.time.Duration.Companion.seconds
2020
import kotlinx.coroutines.*
2121
import kotlinx.coroutines.test.runTest
22+
import org.assertj.core.api.Assertions.assertThat
2223
import org.awaitility.kotlin.await
23-
import org.awaitility.kotlin.until
24+
import org.awaitility.kotlin.withAlias
2425
import org.junit.jupiter.api.extension.RegisterExtension
2526
import org.junit.jupiter.params.ParameterizedTest
2627
import org.junit.jupiter.params.provider.EnumSource
@@ -49,24 +50,33 @@ class CancelInvocation {
4950
val cancelTestClient = CancelTestRunnerClient.fromClient(ingressClient, key)
5051
val blockingServiceClient = CancelTestBlockingServiceClient.fromClient(ingressClient, key)
5152

52-
val id = cancelTestClient.send().startTest(blockingOperation).invocationId
53+
val id =
54+
cancelTestClient.send().startTest(blockingOperation, idempotentCallOptions()).invocationId
5355

5456
val awakeableHolderClient = AwakeableHolderClient.fromClient(ingressClient, "cancel")
55-
56-
await until { runBlocking { awakeableHolderClient.hasAwakeable() } }
57-
58-
awakeableHolderClient.unlock("cancel")
57+
await withAlias
58+
"awakeable is registered" untilAsserted
59+
{
60+
assertThat(awakeableHolderClient.hasAwakeable()).isTrue()
61+
}
62+
awakeableHolderClient.unlock("cancel", idempotentCallOptions())
5963

6064
val client = InvocationApi(ApiClient().setHost(metaURL.host).setPort(metaURL.port))
6165

6266
// The termination signal might arrive before the blocking call to the cancel singleton was
6367
// made, so we need to retry.
64-
await.ignoreException(TimeoutCancellationException::class.java).until {
65-
client.terminateInvocation(id, TerminationMode.CANCEL)
66-
runBlocking { withTimeout(1.seconds) { cancelTestClient.verifyTest() } }
67-
}
68+
await.ignoreException(TimeoutCancellationException::class.java) withAlias
69+
"verify test" untilAsserted
70+
{
71+
client.terminateInvocation(id, TerminationMode.CANCEL)
72+
withTimeout(1.seconds) { cancelTestClient.verifyTest() }
73+
}
6874

6975
// Check that the singleton service is unlocked
70-
blockingServiceClient.isUnlocked()
76+
await withAlias
77+
"blocking service is unlocked" untilAsserted
78+
{
79+
blockingServiceClient.isUnlocked()
80+
}
7181
}
7282
}

‎src/main/kotlin/dev/restate/sdktesting/tests/Ingress.kt‎

Lines changed: 20 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -20,14 +20,12 @@ import dev.restate.sdktesting.infra.*
2020
import java.net.URL
2121
import java.util.*
2222
import java.util.concurrent.TimeUnit
23-
import kotlinx.coroutines.runBlocking
2423
import kotlinx.coroutines.test.runTest
2524
import kotlinx.serialization.encodeToString
2625
import kotlinx.serialization.json.Json
2726
import org.assertj.core.api.Assertions.assertThat
2827
import org.awaitility.kotlin.await
29-
import org.awaitility.kotlin.until
30-
import org.awaitility.kotlin.untilAsserted
28+
import org.awaitility.kotlin.withAlias
3129
import org.junit.jupiter.api.Disabled
3230
import org.junit.jupiter.api.DisplayName
3331
import org.junit.jupiter.api.Test
@@ -86,17 +84,20 @@ class Ingress {
8684

8785
// Await until the idempotency id is cleaned up and the next idempotency call updates the
8886
// counter again
89-
await untilAsserted
87+
await withAlias
88+
"cleanup of the previous idempotent request" untilAsserted
9089
{
91-
runBlocking {
92-
assertThat(counterClient.add(2, requestOptions))
93-
.returns(2, CounterUpdateResponse::oldValue)
94-
.returns(4, CounterUpdateResponse::newValue)
95-
}
90+
assertThat(counterClient.add(2, requestOptions))
91+
.returns(2, CounterUpdateResponse::oldValue)
92+
.returns(4, CounterUpdateResponse::newValue)
9693
}
9794

9895
// State in the counter service is now equal to 4
99-
assertThat(counterClient.get()).isEqualTo(4L)
96+
await withAlias
97+
"Get returns 4 now" untilAsserted
98+
{
99+
assertThat(counterClient.get()).isEqualTo(4L)
100+
}
100101
}
101102

102103
@Test
@@ -128,10 +129,10 @@ class Ingress {
128129
requestOptions)
129130

130131
// Wait for get
131-
await untilAsserted { runBlocking { assertThat(counterClient.get()).isEqualTo(2) } }
132+
await untilAsserted { assertThat(counterClient.get()).isEqualTo(2) }
132133

133134
// Without request options this should be executed immediately and return 4
134-
assertThat(counterClient.add(2))
135+
assertThat(counterClient.add(2, idempotentCallOptions()))
135136
.returns(2, CounterUpdateResponse::oldValue)
136137
.returns(4, CounterUpdateResponse::newValue)
137138
}
@@ -159,10 +160,10 @@ class Ingress {
159160
.isEqualTo(secondInvocationSendStatus.invocationId)
160161

161162
// Wait for get
162-
await untilAsserted { runBlocking { assertThat(counterClient.get()).isEqualTo(2) } }
163+
await untilAsserted { assertThat(counterClient.get()).isEqualTo(2) }
163164

164165
// Without request options this should be executed immediately and return 4
165-
assertThat(counterClient.add(2))
166+
assertThat(counterClient.add(2, idempotentCallOptions()))
166167
.returns(2, CounterUpdateResponse::oldValue)
167168
.returns(4, CounterUpdateResponse::newValue)
168169
}
@@ -201,8 +202,8 @@ class Ingress {
201202

202203
// Unblock
203204
val awakeableHolderClient = AwakeableHolderClient.fromClient(ingressClient, awakeableKey)
204-
await until { runBlocking { awakeableHolderClient.hasAwakeable() } }
205-
awakeableHolderClient.unlock(response)
205+
await untilAsserted { assertThat(awakeableHolderClient.hasAwakeable()).isTrue }
206+
awakeableHolderClient.unlock(response, idempotentCallOptions())
206207

207208
// Attach should be completed
208209
assertThat(blockedFut.get()).isEqualTo(AwakeableResultResponse(response))
@@ -248,8 +249,8 @@ class Ingress {
248249

249250
// Unblock
250251
val awakeableHolderClient = AwakeableHolderClient.fromClient(ingressClient, awakeableKey)
251-
await until { runBlocking { awakeableHolderClient.hasAwakeable() } }
252-
awakeableHolderClient.unlock(response)
252+
await untilAsserted { assertThat(awakeableHolderClient.hasAwakeable()).isTrue }
253+
awakeableHolderClient.unlock(response, idempotentCallOptions())
253254

254255
// Attach should be completed
255256
assertThat(blockedFut.get()).isEqualTo(AwakeableResultResponse(response))
@@ -267,7 +268,7 @@ class Ingress {
267268

268269
assertThat(
269270
TestUtilsServiceClient.fromClient(ingressClient)
270-
.echoHeaders(CallRequestOptions().withHeader(headerName, headerValue)))
271+
.echoHeaders(idempotentCallOptions().withHeader(headerName, headerValue)))
271272
.containsEntry(headerName, headerValue)
272273
}
273274
}

‎src/main/kotlin/dev/restate/sdktesting/tests/KafkaIngress.kt‎

Lines changed: 8 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -22,16 +22,15 @@ import dev.restate.sdktesting.infra.runtimeconfig.KafkaClusterOptions
2222
import dev.restate.sdktesting.infra.runtimeconfig.RestateConfigSchema
2323
import java.net.URL
2424
import java.util.*
25-
import kotlinx.coroutines.runBlocking
2625
import kotlinx.coroutines.test.runTest
2726
import kotlinx.serialization.encodeToString
2827
import kotlinx.serialization.json.Json
2928
import org.apache.kafka.clients.producer.KafkaProducer
3029
import org.apache.kafka.clients.producer.Producer
3130
import org.apache.kafka.clients.producer.ProducerRecord
31+
import org.assertj.core.api.Assertions.assertThat
3232
import org.awaitility.kotlin.await
33-
import org.awaitility.kotlin.matches
34-
import org.awaitility.kotlin.untilCallTo
33+
import org.awaitility.kotlin.withAlias
3534
import org.junit.jupiter.api.Test
3635
import org.junit.jupiter.api.extension.RegisterExtension
3736
import org.junit.jupiter.api.parallel.Execution
@@ -88,13 +87,10 @@ class KafkaIngress {
8887
COUNTER_TOPIC,
8988
listOf(counter to "1", counter to "2", counter to "3"))
9089

91-
// Now wait for the update to be visible
92-
await untilCallTo
90+
await withAlias
91+
"Updates from Kafka are visible in the counter" untilAsserted
9392
{
94-
runBlocking { CounterClient.fromClient(ingressClient, counter).get() }
95-
} matches
96-
{ num ->
97-
num!! == 6L
93+
assertThat(CounterClient.fromClient(ingressClient, counter).get()).isEqualTo(6L)
9894
}
9995
}
10096

@@ -144,13 +140,10 @@ class KafkaIngress {
144140
Json.encodeToString(3).encodeToByteArray())),
145141
))
146142

147-
// Now wait for the update to be visible
148-
await untilCallTo
143+
await withAlias
144+
"Updates from Kafka are visible in the counter" untilAsserted
149145
{
150-
runBlocking { CounterClient.fromClient(ingressClient, counter).get() }
151-
} matches
152-
{ num ->
153-
num!! == 6L
146+
assertThat(CounterClient.fromClient(ingressClient, counter).get()).isEqualTo(6L)
154147
}
155148
}
156149
}

‎src/main/kotlin/dev/restate/sdktesting/tests/KillInvocation.kt‎

Lines changed: 13 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -15,10 +15,10 @@ import dev.restate.sdk.client.Client
1515
import dev.restate.sdktesting.contracts.*
1616
import dev.restate.sdktesting.infra.*
1717
import java.net.URL
18-
import kotlinx.coroutines.runBlocking
1918
import kotlinx.coroutines.test.runTest
19+
import org.assertj.core.api.Assertions.assertThat
2020
import org.awaitility.kotlin.await
21-
import org.awaitility.kotlin.until
21+
import org.awaitility.kotlin.withAlias
2222
import org.junit.jupiter.api.Test
2323
import org.junit.jupiter.api.extension.RegisterExtension
2424

@@ -40,17 +40,22 @@ class KillInvocation {
4040
fun kill(@InjectClient ingressClient: Client, @InjectMetaURL metaURL: URL) = runTest {
4141
val id = KillTestRunnerClient.fromClient(ingressClient).send().startCallTree().invocationId
4242
val awakeableHolderClient = AwakeableHolderClient.fromClient(ingressClient, "kill")
43-
44-
// Await until AwakeableHolder has an awakeable and then complete it.
4543
// With this synchronization point we make sure the call tree has been built before killing it.
46-
await until { runBlocking { awakeableHolderClient.hasAwakeable() } }
47-
awakeableHolderClient.unlock("")
44+
await withAlias
45+
"awakeable is registered" untilAsserted
46+
{
47+
assertThat(awakeableHolderClient.hasAwakeable()).isTrue()
48+
}
49+
awakeableHolderClient.unlock("cancel", idempotentCallOptions())
4850

4951
// Kill the invocation
5052
val client = InvocationApi(ApiClient().setHost(metaURL.host).setPort(metaURL.port))
5153
client.terminateInvocation(id, TerminationMode.KILL)
5254

53-
// Check that the singleton service is unlocked after killing the call tree
54-
KillTestSingletonClient.fromClient(ingressClient, "").isUnlocked()
55+
await withAlias
56+
"singleton service is unlocked after killing the call tree" untilAsserted
57+
{
58+
KillTestSingletonClient.fromClient(ingressClient, "").isUnlocked()
59+
}
5560
}
5661
}

‎src/main/kotlin/dev/restate/sdktesting/tests/KillRuntime.kt‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ class KillRuntime {
4040
) = runTest {
4141
var counterClient = CounterClient.fromClient(ingressClient, "my-key")
4242

43-
val res1 = counterClient.add(1)
43+
val res1 = counterClient.add(1, idempotentCallOptions())
4444
assertThat(res1.oldValue).isEqualTo(0)
4545
assertThat(res1.newValue).isEqualTo(1)
4646

@@ -52,7 +52,7 @@ class KillRuntime {
5252
counterClient =
5353
CounterClient.fromClient(
5454
Client.connect("http://127.0.0.1:${runtimeHandle.getMappedPort(8080)!!}"), "my-key")
55-
val res2 = counterClient.add(2)
55+
val res2 = counterClient.add(2, idempotentCallOptions())
5656
assertThat(res2.oldValue).isEqualTo(1)
5757
assertThat(res2.newValue).isEqualTo(3)
5858
}

‎src/main/kotlin/dev/restate/sdktesting/tests/NonDeterminismErrors.kt‎

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@ import dev.restate.sdktesting.contracts.NonDeterministicDefinitions
1717
import dev.restate.sdktesting.infra.*
1818
import kotlinx.coroutines.test.runTest
1919
import org.assertj.core.api.Assertions.*
20+
import org.awaitility.kotlin.await
21+
import org.awaitility.kotlin.withAlias
2022
import org.junit.jupiter.api.Tag
2123
import org.junit.jupiter.api.extension.RegisterExtension
2224
import org.junit.jupiter.api.parallel.Execution
@@ -51,19 +53,19 @@ class NonDeterminismErrors {
5153
fun method(handlerName: String, @InjectClient ingressClient: Client) = runTest {
5254
// Increment the count first, this makes sure that the counter service is there.
5355
val c = CounterClient.fromClient(ingressClient, handlerName)
54-
c.add(1)
56+
c.add(1, idempotentCallOptions())
5557

5658
assertThatThrownBy {
5759
ingressClient.call(
5860
Target.virtualObject(
5961
NonDeterministicDefinitions.SERVICE_NAME, handlerName, handlerName),
6062
Serde.VOID,
6163
Serde.VOID,
62-
null)
64+
null,
65+
idempotentCallOptions())
6366
}
6467
.isNotNull()
6568

66-
// Assert the counter was not incremented
67-
assertThat(CounterClient.fromClient(ingressClient, handlerName).get()).isEqualTo(1)
69+
await withAlias "counter was not incremented" untilAsserted { assertThat(c.get()).isEqualTo(1) }
6870
}
6971
}

‎src/main/kotlin/dev/restate/sdktesting/tests/PrivateService.kt‎

Lines changed: 17 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ import dev.restate.sdktesting.contracts.*
1717
import dev.restate.sdktesting.infra.*
1818
import java.net.URL
1919
import java.util.*
20+
import kotlinx.coroutines.currentCoroutineContext
2021
import kotlinx.coroutines.runBlocking
2122
import kotlinx.coroutines.test.runTest
2223
import kotlinx.serialization.encodeToString
@@ -25,7 +26,8 @@ import org.assertj.core.api.Assertions.assertThat
2526
import org.assertj.core.api.Assertions.assertThatThrownBy
2627
import org.assertj.core.api.InstanceOfAssertFactories
2728
import org.awaitility.kotlin.await
28-
import org.awaitility.kotlin.untilAsserted
29+
import org.awaitility.kotlin.withAlias
30+
import org.junit.jupiter.api.DisplayName
2931
import org.junit.jupiter.api.Test
3032
import org.junit.jupiter.api.extension.RegisterExtension
3133

@@ -42,6 +44,8 @@ class PrivateService {
4244
}
4345

4446
@Test
47+
@DisplayName(
48+
"Make a handler ingress private and try to call it both directly and through a proxy service")
4549
fun privateService(
4650
@InjectMetaURL metaURL: URL,
4751
@InjectClient ingressClient: Client,
@@ -50,16 +54,18 @@ class PrivateService {
5054
val counterId = UUID.randomUUID().toString()
5155
val counterClient = CounterClient.fromClient(ingressClient, counterId)
5256

53-
counterClient.add(1)
57+
counterClient.add(1, idempotentCallOptions())
5458

5559
// Make the service private
5660
adminServiceClient.modifyService(
5761
CounterDefinitions.SERVICE_NAME, ModifyServiceRequest()._public(false))
5862

5963
// Wait for the service to be private
60-
await untilAsserted
64+
await withAlias
65+
"the service becomes private" untilAsserted
6166
{
62-
assertThatThrownBy { runBlocking { counterClient.get() } }
67+
val ctx = currentCoroutineContext()
68+
assertThatThrownBy { runBlocking(ctx) { counterClient.get() } }
6369
.asInstanceOf(InstanceOfAssertFactories.type(IngressException::class.java))
6470
.returns(400, IngressException::getStatusCode)
6571
}
@@ -71,13 +77,18 @@ class PrivateService {
7177
CounterDefinitions.SERVICE_NAME,
7278
counterId,
7379
"add",
74-
Json.encodeToString(1).encodeToByteArray()))
80+
Json.encodeToString(1).encodeToByteArray()),
81+
idempotentCallOptions())
7582

7683
// Make the service public again
7784
adminServiceClient.modifyService(
7885
CounterDefinitions.SERVICE_NAME, ModifyServiceRequest()._public(true))
7986

8087
// Wait to get the correct count
81-
await untilAsserted { runBlocking { assertThat(counterClient.get()).isEqualTo(2L) } }
88+
await withAlias
89+
"the service becomes public again" untilAsserted
90+
{
91+
assertThat(counterClient.get()).isEqualTo(2L)
92+
}
8293
}
8394
}

0 commit comments

Comments
 (0)