fix: improve observer stability, performance, and security

Signed-off-by: Twiineenock <Twiineenock@users.noreply.github.com>

Author: Twiineenock

.github/workflows/codeql.yml

@@ -54,6 +54,11 @@ jobs:
         # If you are analyzing a compiled language, you can modify the 'build-mode' for that language to customize how
         # your codebase is analyzed, see https://docs.github.com/en/code-security/code-scanning/creating-an-advanced-setup-for-code-scanning/codeql-code-scanning-for-compiled-languages
     steps:
+    - name: Harden Runner
+      uses: step-security/harden-runner@v2.10.1
+      with:
+        egress-policy: audit
+
     - name: Checkout repository
       uses: actions/checkout@v6.0.2
 

.github/workflows/maven.yml

@@ -20,6 +20,11 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
+      - name: Harden Runner
+        uses: step-security/harden-runner@v2.10.1
+        with:
+          egress-policy: audit
+
       - name: Checkout Repo
         uses: actions/checkout@v6.0.2
 

.github/workflows/scorecard.yml

@@ -31,6 +31,11 @@ jobs:
       # actions: read
 
     steps:
+      - name: Harden Runner
+        uses: step-security/harden-runner@v2.10.1
+        with:
+          egress-policy: audit
+
       - name: "Checkout code"
         uses: actions/checkout@0c366fd6a839edf440554fa01a7085ccba70ac98 # v5.0.1
         with:

hiero-enterprise-base/src/main/java/org/hiero/base/observer/AbstractPollingObserver.java

@@ -29,17 +29,17 @@ public abstract class AbstractPollingObserver<T> {
     /**
      * Creates a new polling observer.
      *
+     * @param executorService The shared executor service to use for polling.
      * @param pollingInterval How often to check for new events.
      * @param listener The callback to trigger when an event is found.
      */
-    protected AbstractPollingObserver(@NonNull Duration pollingInterval, @NonNull EventObserver<T> listener) {
+    protected AbstractPollingObserver(
+            @NonNull ScheduledExecutorService executorService,
+            @NonNull Duration pollingInterval,
+            @NonNull EventObserver<T> listener) {
+        this.executorService = executorService;
         this.pollingInterval = pollingInterval;
         this.listener = listener;
-        this.executorService = Executors.newSingleThreadScheduledExecutor(r -> {
-            Thread thread = new Thread(r, "hiero-observer-" + getClass().getSimpleName());
-            thread.setDaemon(true);
-            return thread;
-        });
     }
 
     /**
@@ -48,34 +48,43 @@ protected AbstractPollingObserver(@NonNull Duration pollingInterval, @NonNull Ev
     public void start() {
         if (running.compareAndSet(false, true)) {
             log.info("Starting Hiero observer: {}", getClass().getSimpleName());
-            executorService.scheduleAtFixedRate(this::pollSafe, 0, pollingInterval.toMillis(), TimeUnit.MILLISECONDS);
+            scheduleNext(0);
+        }
+    }
+
+    private void scheduleNext(long delayMs) {
+        if (running.get()) {
+            executorService.schedule(this::pollSafe, delayMs, TimeUnit.MILLISECONDS);
         }
     }
 
     /**
      * Stops the background polling task.
      */
     public void stop() {
-        if (running.compareAndSet(true, false)) {
-            log.info("Stopping Hiero observer: {}", getClass().getSimpleName());
-            executorService.shutdown();
-        }
+        running.set(false);
+        log.info("Stopping Hiero observer: {}", getClass().getSimpleName());
     }
 
     private void pollSafe() {
         try {
-            poll();
+            boolean hasMore = poll();
+            // If there's more data (pagination), poll again immediately.
+            // Otherwise, wait for the polling interval.
+            scheduleNext(hasMore ? 0 : pollingInterval.toMillis());
         } catch (Exception e) {
             log.error("Unexpected error during Hiero polling in {}", getClass().getSimpleName(), e);
+            scheduleNext(pollingInterval.toMillis());
         }
     }
 
     /**
      * The actual polling logic to be implemented by subclasses.
      *
+     * @return true if there is more data to poll immediately (pagination), false otherwise.
      * @throws Exception if polling fails.
      */
-    public abstract void poll() throws Exception;
+    public abstract boolean poll() throws Exception;
 
     /**
      * Notifies the listener of a new event.

hiero-enterprise-base/src/main/java/org/hiero/base/observer/TopicObserver.java

@@ -5,6 +5,7 @@
 import java.time.Instant;
 import java.util.Comparator;
 import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
 import org.hiero.base.data.Page;
 import org.hiero.base.data.TopicMessage;
 import org.hiero.base.mirrornode.TopicRepository;
@@ -26,29 +27,36 @@ public class TopicObserver extends AbstractPollingObserver<TopicMessage> {
     /**
      * Creates a new topic observer.
      *
+     * @param executorService The shared executor service to use for polling.
      * @param repository The repository to use for polling.
      * @param topicId The topic to monitor.
      * @param pollingInterval How often to poll the mirror node.
      * @param listener The callback for new messages.
      */
     public TopicObserver(
+            @NonNull ScheduledExecutorService executorService,
             @NonNull TopicRepository repository,
             @NonNull TopicId topicId,
             @NonNull Duration pollingInterval,
             @NonNull EventObserver<TopicMessage> listener) {
-        super(pollingInterval, listener);
+        super(executorService, pollingInterval, listener);
         this.repository = repository;
         this.topicId = topicId;
         this.lastSeenTimestamp = Instant.now();
     }
 
     @Override
-    public void poll() throws Exception {
+    public boolean poll() throws Exception {
         log.trace("Polling messages for topic {} after {}", topicId, lastSeenTimestamp);
 
         Page<TopicMessage> page = repository.getMessages(topicId, lastSeenTimestamp);
-        List<TopicMessage> messages = new java.util.ArrayList<>(page.getData());
+        processPage(page);
+
+        return page.hasNext();
+    }
 
+    private void processPage(Page<TopicMessage> page) {
+        List<TopicMessage> messages = new java.util.ArrayList<>(page.getData());
         if (messages.isEmpty()) {
             return;
         }

hiero-enterprise-base/src/main/java/org/hiero/base/observer/TransactionObserver.java

@@ -5,6 +5,7 @@
 import java.time.Instant;
 import java.util.Comparator;
 import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
 import org.hiero.base.data.Page;
 import org.hiero.base.data.TransactionInfo;
 import org.hiero.base.mirrornode.TransactionRepository;
@@ -29,30 +30,37 @@ public class TransactionObserver extends AbstractPollingObserver<TransactionInfo
     /**
      * Creates a new transaction observer.
      *
+     * @param executorService The shared executor service to use for polling.
      * @param repository The repository to use for polling.
      * @param accountId The account to monitor.
      * @param pollingInterval How often to poll the mirror node.
      * @param listener The callback for new transactions.
      */
     public TransactionObserver(
+            @NonNull ScheduledExecutorService executorService,
             @NonNull TransactionRepository repository,
             @NonNull AccountId accountId,
             @NonNull Duration pollingInterval,
             @NonNull EventObserver<TransactionInfo> listener) {
-        super(pollingInterval, listener);
+        super(executorService, pollingInterval, listener);
         this.repository = repository;
         this.accountId = accountId;
         // Start from current time to avoid processing historical transactions by default
         this.lastSeenTimestamp = Instant.now();
     }
 
     @Override
-    public void poll() throws Exception {
+    public boolean poll() throws Exception {
         log.trace("Polling transactions for account {} after {}", accountId, lastSeenTimestamp);
 
         Page<TransactionInfo> page = repository.findByAccount(accountId, lastSeenTimestamp);
-        List<TransactionInfo> transactions = new java.util.ArrayList<>(page.getData());
+        processPage(page);
+
+        return page.hasNext();
+    }
 
+    private void processPage(Page<TransactionInfo> page) {
+        List<TransactionInfo> transactions = new java.util.ArrayList<>(page.getData());
         if (transactions.isEmpty()) {
             return;
         }

hiero-enterprise-base/src/test/java/org/hiero/base/test/TransactionObserverTest.java

@@ -9,6 +9,8 @@
 import java.time.Instant;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import org.hiero.base.data.Page;
 import org.hiero.base.data.TransactionInfo;
 import org.hiero.base.mirrornode.TransactionRepository;
@@ -30,11 +32,12 @@ class TransactionObserverTest {
 
     private TransactionObserver observer;
     private final AccountId accountId = AccountId.fromString("0.0.1234");
+    private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
 
     @BeforeEach
     void setUp() {
         MockitoAnnotations.openMocks(this);
-        observer = new TransactionObserver(repository, accountId, Duration.ofMillis(100), listener);
+        observer = new TransactionObserver(executorService, repository, accountId, Duration.ofMillis(100), listener);
     }
 
     @Test

hiero-enterprise-spring-sample/src/main/java/org/hiero/spring/sample/HieroEventListener.java

@@ -19,8 +19,8 @@ public class HieroEventListener {
      */
     @HieroTransactionListener(account = "0.0.1234", interval = 10000)
     public void onTransaction(TransactionInfo tx) {
-        log.info(">>> [Hiero Event] New transaction detected: {} (Status: {})", 
-                 tx.transactionId(), tx.status());
+        log.info(">>> [Hiero Event] New transaction detected: {} (Result: {})", 
+                 tx.transactionId(), tx.result());
     }
 
     /**

hiero-enterprise-spring/src/main/java/org/hiero/spring/implementation/HieroAutoConfiguration.java

@@ -129,9 +129,9 @@ MirrorNodeClient mirrorNodeClient(final HieroContext hieroContext) {
       final String mirrorNodeEndpointProtocol = url.getProtocol();
       final String mirrorNodeEndpointHost = url.getHost();
       final int mirrorNodeEndpointPort;
-      if (mirrorNodeEndpointProtocol == "https" && url.getPort() == -1) {
+      if (mirrorNodeEndpointProtocol.equals("https") && url.getPort() == -1) {
         mirrorNodeEndpointPort = 443;
-      } else if (mirrorNodeEndpointProtocol == "http" && url.getPort() == -1) {
+      } else if (mirrorNodeEndpointProtocol.equals("http") && url.getPort() == -1) {
         mirrorNodeEndpointPort = 80;
       } else if (url.getPort() == -1) {
         mirrorNodeEndpointPort = 443;

hiero-enterprise-spring/src/main/java/org/hiero/spring/implementation/HieroListenerProcessor.java

@@ -15,9 +15,12 @@
 import org.hiero.spring.annotation.HieroTransactionListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.aop.framework.AopProxyUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.context.SmartLifecycle;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
 import org.springframework.util.ReflectionUtils;
 
 /**
@@ -30,17 +33,29 @@ public class HieroListenerProcessor implements BeanPostProcessor, SmartLifecycle
 
     private final TransactionRepository transactionRepository;
     private final TopicRepository topicRepository;
+    private final TaskScheduler taskScheduler;
     private final List<AbstractPollingObserver<?>> observers = new ArrayList<>();
     private boolean running = false;
 
     public HieroListenerProcessor(TransactionRepository transactionRepository, TopicRepository topicRepository) {
         this.transactionRepository = transactionRepository;
         this.topicRepository = topicRepository;
+        this.taskScheduler = createDefaultTaskScheduler();
+    }
+
+    private TaskScheduler createDefaultTaskScheduler() {
+        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+        scheduler.setPoolSize(Runtime.getRuntime().availableProcessors());
+        scheduler.setThreadNamePrefix("hiero-observer-");
+        scheduler.setDaemon(true);
+        scheduler.initialize();
+        return scheduler;
     }
 
     @Override
     public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
-        ReflectionUtils.doWithMethods(bean.getClass(), method -> {
+        Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
+        ReflectionUtils.doWithMethods(targetClass, method -> {
             if (method.isAnnotationPresent(HieroTransactionListener.class)) {
                 processTransactionListener(bean, method);
             }
@@ -56,6 +71,7 @@ private void processTransactionListener(Object bean, Method method) {
         log.info("Registering transaction listener for account {} on method {}", annotation.account(), method.getName());
 
         TransactionObserver observer = new TransactionObserver(
+                ((ThreadPoolTaskScheduler) taskScheduler).getScheduledExecutor(),
                 transactionRepository,
                 AccountId.fromString(annotation.account()),
                 Duration.ofMillis(annotation.interval()),
@@ -72,6 +88,7 @@ private void processTopicListener(Object bean, Method method) {
         log.info("Registering topic listener for topic {} on method {}", annotation.topicId(), method.getName());
 
         TopicObserver observer = new TopicObserver(
+                ((ThreadPoolTaskScheduler) taskScheduler).getScheduledExecutor(),
                 topicRepository,
                 TopicId.fromString(annotation.topicId()),
                 Duration.ofMillis(annotation.interval()),