feat(pool): add StateMachine realizations

tarantool-core/pom.xml

@@ -72,6 +72,11 @@
       <groupId>io.tarantool</groupId>
       <artifactId>testcontainers</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-junit-jupiter</artifactId>
+      <scope>test</scope>
+    </dependency>
     <!-- End test -->
   </dependencies>
 </project>

tarantool-core/src/main/java/io/tarantool/core/state/AbstractDrainLoopMachine.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractDrainLoopMachine<E extends Event> implements StateMachine<E> {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(AbstractDrainLoopMachine.class);
+
+  private final Queue<E> events = new ConcurrentLinkedQueue<>();
+  private final AtomicInteger wip = new AtomicInteger(0);
+
+  @Override
+  public void handle(E event) {
+    this.events.offer(event);
+
+    if (this.wip.getAndIncrement() != 0) {
+      return;
+    }
+
+    int missed = 1;
+    try {
+      do {
+        E e;
+        while ((e = this.events.poll()) != null) {
+          try {
+            processEvent(e);
+          } catch (Throwable ex) {
+            LOGGER.error("[{}]: unexpected error in event handler. Try handle error...", this, ex);
+            try {
+              onHandleError(ex);
+            } catch (Throwable failSafeEx) {
+              LOGGER.error("[{}]: fail-safe handler failed", this, failSafeEx);
+            }
+          }
+        }
+        missed = this.wip.addAndGet(-missed);
+      } while (missed != 0);
+    } catch (Throwable criticalJvmError) {
+      this.wip.set(0);
+      throw criticalJvmError;
+    }
+  }
+
+  protected abstract void processEvent(E event);
+
+  protected abstract void onHandleError(Throwable error);
+}

tarantool-core/src/main/java/io/tarantool/core/state/AbstractNettyMachine.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+import java.util.concurrent.RejectedExecutionException;
+
+import io.netty.channel.EventLoop;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractNettyMachine<E extends Event> implements StateMachine<E> {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(AbstractNettyMachine.class);
+
+  protected final EventLoop loop;
+
+  protected AbstractNettyMachine(EventLoop loop) {
+    this.loop = loop;
+  }
+
+  @Override
+  public void handle(E event) {
+    if (this.loop.inEventLoop()) {
+      internalHandle(event);
+    } else {
+      try {
+        this.loop.execute(() -> internalHandle(event));
+      } catch (RejectedExecutionException ex) {
+        LOGGER.error(
+            "[{}]: catch an unexpected machine event execution. Try handle this...", this, ex);
+        try {
+          onHandleRejectedException(event, ex);
+        } catch (Throwable failSafeEx) {
+          LOGGER.error("[{}]: fail-safe handler failed", this, failSafeEx);
+        }
+      }
+    }
+  }
+
+  private void internalHandle(E event) {
+    try {
+      processEvent(event);
+    } catch (Throwable error) {
+      try {
+        LOGGER.error("[{}]: unexpected error in event handler. Try handle error...", this, error);
+        onHandleError(error);
+      } catch (Throwable failSafeEx) {
+        LOGGER.error("[{}]: fail-safe handler failed", this, failSafeEx);
+      }
+    }
+  }
+
+  protected abstract void processEvent(E event);
+
+  protected abstract void onHandleError(Throwable error);
+
+  protected abstract void onHandleRejectedException(E event, RejectedExecutionException e);
+}

tarantool-core/src/main/java/io/tarantool/core/state/Event.java

@@ -0,0 +1,8 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+public interface Event {}

tarantool-core/src/main/java/io/tarantool/core/state/State.java

@@ -0,0 +1,11 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+public interface State<E extends Event> {
+
+  void handle(E event);
+}

tarantool-core/src/main/java/io/tarantool/core/state/StateMachine.java

@@ -0,0 +1,11 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+public interface StateMachine<E extends Event> {
+
+  void handle(E event);
+}

tarantool-core/src/test/java/io/tarantool/core/state/AbstractNettyMachineTest.java

@@ -0,0 +1,151 @@
+/*
+ * Copyright (c) 2026 VK DIGITAL TECHNOLOGIES LIMITED LIABILITY COMPANY
+ * All Rights Reserved.
+ */
+
+package io.tarantool.core.state;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.RejectedExecutionException;
+
+import io.netty.channel.EventLoop;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+class AbstractNettyMachineTest {
+
+  private EventLoop mockLoop;
+  private TestNettyMachine machine;
+
+  @BeforeEach
+  void setUp() {
+    mockLoop = Mockito.mock(EventLoop.class);
+    machine = new TestNettyMachine(mockLoop);
+  }
+
+  @Test
+  void testHandleInEventLoop() {
+    Mockito.when(mockLoop.inEventLoop()).thenReturn(true);
+    final Event event = Mockito.mock(Event.class);
+
+    machine.handle(event);
+
+    Assertions.assertEquals(1, machine.processedEvents.size());
+    Assertions.assertSame(event, machine.processedEvents.get(0));
+    Mockito.verify(mockLoop, Mockito.never()).execute(Mockito.any(Runnable.class));
+  }
+
+  @Test
+  void testHandleOutsideEventLoop() {
+    Mockito.when(mockLoop.inEventLoop()).thenReturn(false);
+    final Event event = Mockito.mock(Event.class);
+
+    machine.handle(event);
+
+    Assertions.assertEquals(0, machine.processedEvents.size());
+
+    final ArgumentCaptor<Runnable> runnableCaptor = ArgumentCaptor.forClass(Runnable.class);
+    Mockito.verify(mockLoop, Mockito.times(1)).execute(runnableCaptor.capture());
+
+    runnableCaptor.getValue().run();
+
+    Assertions.assertEquals(1, machine.processedEvents.size());
+    Assertions.assertSame(event, machine.processedEvents.get(0));
+  }
+
+  @Test
+  void testRecoveryAfterThrowable() {
+    Mockito.when(mockLoop.inEventLoop()).thenReturn(true);
+    final Event event = Mockito.mock(Event.class);
+    final RuntimeException expectedException = new RuntimeException("Сбой бизнес-логики");
+
+    machine.onProcessHook =
+        () -> {
+          throw expectedException;
+        };
+
+    Assertions.assertDoesNotThrow(() -> machine.handle(event));
+
+    Assertions.assertEquals(1, machine.caughtErrors.size());
+    Assertions.assertSame(expectedException, machine.caughtErrors.get(0));
+    Assertions.assertEquals(0, machine.processedEvents.size());
+  }
+
+  @Test
+  void testHandleRejectedExecutionException() {
+    Mockito.when(mockLoop.inEventLoop()).thenReturn(false);
+    Mockito.doThrow(new RejectedExecutionException("Loop is shutdown"))
+        .when(mockLoop)
+        .execute(Mockito.any(Runnable.class));
+
+    final Event event = Mockito.mock(Event.class);
+
+    Assertions.assertDoesNotThrow(() -> machine.handle(event));
+
+    // Проверяем, что сработал наш новый безопасный обработчик отказа и контекст события НЕ потерян
+    Assertions.assertEquals(1, machine.rejectedEvents.size());
+    Assertions.assertSame(event, machine.rejectedEvents.get(0));
+    Assertions.assertEquals(0, machine.processedEvents.size());
+  }
+
+  @Test
+  void testFailSafeHandlerProtectsAgainstDoubleFault() {
+    Mockito.when(mockLoop.inEventLoop()).thenReturn(true);
+    final Event event = Mockito.mock(Event.class);
+
+    machine.onProcessHook =
+        () -> {
+          throw new RuntimeException("First error");
+        };
+    TestNettyMachine brokenMachine =
+        new TestNettyMachine(mockLoop) {
+          @Override
+          protected void processEvent(Event event) {
+            throw new RuntimeException("First error");
+          }
+
+          @Override
+          protected void onHandleError(Throwable error) {
+            throw new RuntimeException("Second error onError inside");
+          }
+        };
+
+    Assertions.assertDoesNotThrow(() -> brokenMachine.handle(event));
+  }
+
+  private static class TestNettyMachine extends AbstractNettyMachine<Event> {
+
+    final List<Event> processedEvents = Collections.synchronizedList(new ArrayList<>());
+    final List<Throwable> caughtErrors = Collections.synchronizedList(new ArrayList<>());
+    final List<Event> rejectedEvents = Collections.synchronizedList(new ArrayList<>());
+
+    volatile Runnable onProcessHook = null;
+
+    protected TestNettyMachine(EventLoop loop) {
+      super(loop);
+    }
+
+    @Override
+    protected void processEvent(Event event) {
+      if (onProcessHook != null) {
+        onProcessHook.run();
+      }
+      processedEvents.add(event);
+    }
+
+    @Override
+    protected void onHandleError(Throwable error) {
+      caughtErrors.add(error);
+    }
+
+    @Override
+    protected void onHandleRejectedException(Event event, RejectedExecutionException e) {
+      rejectedEvents.add(event);
+    }
+  }
+}