diff --git a/tests/src/test/kotlin/dev/restate/e2e/runtime/IngressServiceTest.kt b/tests/src/test/kotlin/dev/restate/e2e/runtime/IngressServiceTest.kt index 6db609e1..75018f37 100644 --- a/tests/src/test/kotlin/dev/restate/e2e/runtime/IngressServiceTest.kt +++ b/tests/src/test/kotlin/dev/restate/e2e/runtime/IngressServiceTest.kt @@ -6,23 +6,31 @@ import dev.restate.e2e.Utils.jacksonBodyPublisher import dev.restate.e2e.services.counter.CounterGrpc import dev.restate.e2e.services.counter.CounterGrpc.CounterBlockingStub import dev.restate.e2e.services.counter.CounterProto +import dev.restate.e2e.services.counter.CounterProto.CounterRequest +import dev.restate.e2e.services.counter.CounterProto.CounterUpdateResult +import dev.restate.e2e.services.counter.CounterProto.GetResponse import dev.restate.e2e.services.counter.counterAddRequest +import dev.restate.e2e.services.counter.counterRequest import dev.restate.e2e.utils.InjectBlockingStub import dev.restate.e2e.utils.InjectGrpcIngressURL import dev.restate.e2e.utils.RestateDeployer import dev.restate.e2e.utils.RestateDeployerExtension import dev.restate.generated.IngressGrpc.IngressBlockingStub +import dev.restate.generated.idempotentInvokeRequest import dev.restate.generated.invokeRequest import java.net.URI import java.net.URL import java.net.http.HttpClient import java.net.http.HttpRequest import java.util.UUID +import java.util.concurrent.TimeUnit import org.assertj.core.api.Assertions.assertThat import org.awaitility.kotlin.await import org.awaitility.kotlin.matches +import org.awaitility.kotlin.untilAsserted import org.awaitility.kotlin.untilCallTo import org.junit.jupiter.api.Test +import org.junit.jupiter.api.Timeout import org.junit.jupiter.api.extension.RegisterExtension import org.junit.jupiter.api.parallel.Execution import org.junit.jupiter.api.parallel.ExecutionMode @@ -93,7 +101,7 @@ class IngressServiceTest { invokeRequest { service = CounterGrpc.SERVICE_NAME method = CounterGrpc.getAddMethod().bareMethodName.toString() - argument = + pb = counterAddRequest { counterName = counterRandomName value = 2 @@ -112,4 +120,58 @@ class IngressServiceTest { num!!.value == 2L } } + + @Test + @Execution(ExecutionMode.CONCURRENT) + @Timeout(value = 15, unit = TimeUnit.SECONDS) + fun idempotentInvoke( + @InjectBlockingStub ingressClient: IngressBlockingStub, + @InjectBlockingStub counterClient: CounterBlockingStub + ) { + val counterRandomName = UUID.randomUUID().toString() + val myIdempotencyId = UUID.randomUUID().toString() + val invokeRequest = idempotentInvokeRequest { + idempotencyId = myIdempotencyId + service = CounterGrpc.SERVICE_NAME + method = CounterGrpc.getGetAndAddMethod().bareMethodName.toString() + pb = + counterAddRequest { + counterName = counterRandomName + value = 2 + } + .toByteString() + retentionPeriodSec = 3 + } + + // First call updates the value + val firstResponse = + ingressClient.idempotentInvoke( + invokeRequest) + assertThat(CounterUpdateResult.parseFrom(firstResponse.pb)) + .returns(0, CounterUpdateResult::getOldValue) + .returns(2, CounterUpdateResult::getNewValue) + + // Next call returns the same value + val secondResponse = + ingressClient.idempotentInvoke( + invokeRequest) + assertThat(CounterUpdateResult.parseFrom(secondResponse.pb)) + .returns(0L, CounterUpdateResult::getOldValue) + .returns(2L, CounterUpdateResult::getNewValue) + + // Await until the idempotency id is cleaned up and the next idempotency call updates the + // counter again + await untilAsserted + { + val response = + ingressClient.idempotentInvoke(invokeRequest) + assertThat(CounterUpdateResult.parseFrom(response.pb)) + .returns(2, CounterUpdateResult::getOldValue) + .returns(4, CounterUpdateResult::getNewValue) + } + + // State in the counter service is now equal to 2 + assertThat(counterClient.get(counterRequest { counterName = counterRandomName })) + .returns(4L, GetResponse::getValue) + } } diff --git a/tests/src/test/kotlin/dev/restate/e2e/runtime/KillInvocationTest.kt b/tests/src/test/kotlin/dev/restate/e2e/runtime/KillInvocationTest.kt index 4ef1f0a5..94f08565 100644 --- a/tests/src/test/kotlin/dev/restate/e2e/runtime/KillInvocationTest.kt +++ b/tests/src/test/kotlin/dev/restate/e2e/runtime/KillInvocationTest.kt @@ -53,7 +53,7 @@ class KillInvocationTest { invokeRequest { service = CounterGrpc.SERVICE_NAME method = CounterGrpc.getInfiniteIncrementLoopMethod().bareMethodName!! - argument = counterRequest.toByteString() + pb = counterRequest.toByteString() }) .id diff --git a/tests/src/test/kotlin/dev/restate/e2e/runtime/RetryOnUnknownServiceTest.kt b/tests/src/test/kotlin/dev/restate/e2e/runtime/RetryOnUnknownServiceTest.kt index e5a00b75..92db2743 100644 --- a/tests/src/test/kotlin/dev/restate/e2e/runtime/RetryOnUnknownServiceTest.kt +++ b/tests/src/test/kotlin/dev/restate/e2e/runtime/RetryOnUnknownServiceTest.kt @@ -107,7 +107,7 @@ class RetryOnUnknownServiceTest { invokeRequest { service = ProxyServiceGrpc.SERVICE_NAME method = methodName - argument = request.toByteString() + pb = request.toByteString() }) // Await until we got a try count of 2