createTimerChain was comparing the same timestamp for future validation

Signed-off-by: salaboy <[email protected]>

Author: salaboy

client/src/main/java/io/dapr/durabletask/TaskOrchestrationExecutor.java

@@ -1038,7 +1038,7 @@ public TimerTask(Instant finalFireAt) {
             // if necessary. Otherwise, we return and no more sub-timers are created.
             private CompletableFuture<Void> createTimerChain(Instant finalFireAt, CompletableFuture<Void> currentFuture) {
                 return currentFuture.thenRun(() -> {
-                    if (currentInstant.compareTo(finalFireAt) > 0) {
+                    if (currentInstant.compareTo(finalFireAt) >= 0) {
                         return;
                     }
                     Task<Void> nextTimer = createTimerTask(finalFireAt);

client/src/test/java/io/dapr/durabletask/IntegrationTests.java

@@ -87,8 +87,14 @@ void emptyOrchestration() throws TimeoutException {
     void singleTimer() throws IOException, TimeoutException {
         final String orchestratorName = "SingleTimer";
         final Duration delay = Duration.ofSeconds(3);
+        AtomicReferenceArray<LocalDateTime> timestamps = new AtomicReferenceArray<>(2);
+        AtomicInteger counter = new AtomicInteger();
         DurableTaskGrpcWorker worker = this.createWorkerBuilder()
-            .addOrchestrator(orchestratorName, ctx -> ctx.createTimer(delay).await())
+            .addOrchestrator(orchestratorName, ctx -> {
+                timestamps.set(counter.get(), LocalDateTime.now());
+                counter.incrementAndGet();
+                ctx.createTimer(delay).await();
+            })
             .buildAndStart();
 
         DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
@@ -103,11 +109,22 @@ void singleTimer() throws IOException, TimeoutException {
             long expectedCompletionSecond = instance.getCreatedAt().plus(delay).getEpochSecond();
             long actualCompletionSecond = instance.getLastUpdatedAt().getEpochSecond();
             assertTrue(expectedCompletionSecond <= actualCompletionSecond);
+
+            // Verify that the correct number of timers were created
+            // This should yield 2 (first invocation + replay invocations for internal timers)
+            assertEquals(2, counter.get());
+
+            // Verify that each timer is the expected length
+            int[] secondsElapsed = new int[1];
+            for (int i = 0; i < timestamps.length() - 1; i++) {
+                secondsElapsed[i] = timestamps.get(i + 1).getSecond() - timestamps.get(i).getSecond();
+            }
+            assertEquals(secondsElapsed[0], 3);
+
         }
     }
 
     @Test
-    @Disabled("Test is disabled for investigation, fixing the test retry pattern exposed the failure (could be timer creation issue)")
     void longTimer() throws TimeoutException {
         final String orchestratorName = "LongTimer";
         final Duration delay = Duration.ofSeconds(7);
@@ -285,6 +302,31 @@ void singleTimeStampTimer() throws IOException, TimeoutException {
         }
     }
 
+
+    @Test
+    void singleTimeStampCreateTimer() throws IOException, TimeoutException {
+        final String orchestratorName = "SingleTimeStampTimer";
+        final Duration delay = Duration.ofSeconds(3);
+        final ZonedDateTime zonedDateTime = ZonedDateTime.of(LocalDateTime.now().plusSeconds(delay.getSeconds()), ZoneId.systemDefault());
+        DurableTaskGrpcWorker worker = this.createWorkerBuilder()
+                .addOrchestrator(orchestratorName, ctx -> ctx.createTimer(zonedDateTime).await())
+                .buildAndStart();
+
+        DurableTaskClient client = new DurableTaskGrpcClientBuilder().build();
+        try (worker; client) {
+            String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName);
+            Duration timeout = delay.plus(defaultTimeout);
+            OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, timeout, false);
+            assertNotNull(instance);
+            assertEquals(OrchestrationRuntimeStatus.COMPLETED, instance.getRuntimeStatus());
+
+            // Verify that the delay actually happened
+            long expectedCompletionSecond = zonedDateTime.toInstant().getEpochSecond();
+            long actualCompletionSecond = instance.getLastUpdatedAt().getEpochSecond();
+            assertTrue(expectedCompletionSecond <= actualCompletionSecond);
+        }
+    }
+
     @Test
     void isReplaying() throws IOException, InterruptedException, TimeoutException {
         final String orchestratorName = "SingleTimer";