Remove LiveStreamPublisherManager and make Autonomous greedy configureable

Signed-off-by: Nana Essilfie-Conduah <nana@swirldslabs.com>

Author: Nana-EC

block-node/app/src/main/java/org/hiero/block/node/app/DefaultThreadPoolManager.java

@@ -102,7 +102,8 @@ public ExecutorService createSingleThreadExecutor(
      */
     @NonNull
     @Override
-    public ScheduledExecutorService createSingleThreadScheduledExecutor(
+    public ScheduledExecutorService createVirtualThreadScheduledExecutor(
+            int corePoolSize,
             @Nullable final String threadName,
             @Nullable final Thread.UncaughtExceptionHandler uncaughtExceptionHandler) {
         Thread.Builder factoryBuilder = Thread.ofVirtual();
@@ -111,10 +112,14 @@ public ScheduledExecutorService createSingleThreadScheduledExecutor(
         }
         if (threadName != null) {
             factoryBuilder.name(threadName, 0);
-            return Executors.newSingleThreadScheduledExecutor(factoryBuilder.factory());
+            return corePoolSize <= 1
+                    ? Executors.newSingleThreadScheduledExecutor(factoryBuilder.factory())
+                    : Executors.newScheduledThreadPool(corePoolSize, factoryBuilder.factory());
         } else {
             final ThreadFactory factory = factoryBuilder.factory();
-            return Executors.newSingleThreadScheduledExecutor(factory);
+            return corePoolSize <= 1
+                    ? Executors.newSingleThreadScheduledExecutor(factory)
+                    : Executors.newScheduledThreadPool(corePoolSize, factory);
         }
     }
 }

block-node/app/src/testFixtures/java/org/hiero/block/node/app/fixtures/async/ScheduledBlockingExecutor.java

@@ -3,15 +3,20 @@
 
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 
 public class ScheduledBlockingExecutor extends ScheduledThreadPoolExecutor {
     /** The work queue that will be used to hold the tasks. */
     private final BlockingQueue<Runnable> workQueue;
 
     public ScheduledBlockingExecutor(@NonNull final BlockingQueue<Runnable> workQueue) {
-        super(1, Executors.defaultThreadFactory(), new AbortPolicy());
+        super(1, Thread.ofVirtual().factory(), new AbortPolicy());
+
+        this.workQueue = workQueue; // actual work queue
+    }
+
+    public ScheduledBlockingExecutor(int corePoolSize, @NonNull final BlockingQueue<Runnable> workQueue) {
+        super(corePoolSize, Thread.ofVirtual().factory(), new AbortPolicy());
 
         this.workQueue = workQueue; // actual work queue
     }

block-node/app/src/testFixtures/java/org/hiero/block/node/app/fixtures/async/TestThreadPoolManager.java

@@ -60,8 +60,10 @@ public T createSingleThreadExecutor(
 
     @NonNull
     @Override
-    public ScheduledExecutorService createSingleThreadScheduledExecutor(
-            @Nullable String threadName, @Nullable UncaughtExceptionHandler uncaughtExceptionHandler) {
+    public ScheduledExecutorService createVirtualThreadScheduledExecutor(
+            int corePoolSize,
+            @Nullable String threadName,
+            @Nullable UncaughtExceptionHandler uncaughtExceptionHandler) {
         return scheduledExecutor;
     }
 

block-node/backfill/src/main/java/org/hiero/block/node/backfill/BackfillConfiguration.java

@@ -23,6 +23,7 @@
  *                                  process indefinitely in case something unexpected happens, this would allow for self-recovery
  * @param grpcOverallTimeout single timeout configuration for gRPC Client construction, connectTimeout, readTimeout and pollWaitTime
  * @param enableTLS if enabled will assume block-node client supports tls connection.
+ * @param greedy if enabled will search and retrieve blocks beyond latestAcknowledged to ensure BN doesn't fall too far behind.
  */
 @ConfigData("backfill")
 public record BackfillConfiguration(
@@ -37,4 +38,5 @@ public record BackfillConfiguration(
         @Loggable @ConfigProperty(defaultValue = "15000") @Min(5) int initialDelay,
         @Loggable @ConfigProperty(defaultValue = "1000") @Min(500) int perBlockProcessingTimeout,
         @Loggable @ConfigProperty(defaultValue = "60000") @Min(10000) int grpcOverallTimeout,
-        @Loggable @ConfigProperty(defaultValue = "false") boolean enableTLS) {}
+        @Loggable @ConfigProperty(defaultValue = "false") boolean enableTLS,
+        @Loggable @ConfigProperty(defaultValue = "true") boolean greedy) {}

block-node/backfill/src/main/java/org/hiero/block/node/backfill/BackfillPlugin.java

@@ -61,6 +61,7 @@ public class BackfillPlugin implements BlockNodePlugin, BlockNotificationHandler
     private volatile boolean onDemandError = false;
     private final AtomicLong autonomousBackfillEndBlock = new AtomicLong(-1);
     private final AtomicLong onDemandBackfillStartBlock = new AtomicLong(-1);
+    private final AtomicLong lastAcknowledgedBlockObserved = new AtomicLong(-1);
 
     // Metrics
     private Counter backfillGapsDetected;
@@ -221,7 +222,13 @@ public void start() {
                 "Scheduling backfill process to start in {0} milliseconds",
                 backfillConfiguration.initialDelay());
         scheduler = Executors.newScheduledThreadPool(
-                2); // Two threads: one for autonomous backfill, one for on-demand backfill
+                2,
+                Thread.ofVirtual().factory()); // Two threads: one for autonomous backfill, one for on-demand backfill
+        //        scheduler = context.threadPoolManager().createVirtualThreadScheduledExecutor(
+        //            2,  // Two threads: one for autonomous backfill, one for on-demand backfill
+        //            "BackfillPluginRunner",
+        //            (t, e) -> LOGGER.log(ERROR, "Uncaught exception in thread: " + t.getName(), e));
+
         scheduler.scheduleAtFixedRate(
                 this::detectGaps,
                 backfillConfiguration.initialDelay(),
@@ -257,45 +264,42 @@ private boolean isOnDemandBackfillRunning() {
         return onDemandBackfillStartBlock.get() != -1;
     }
 
-    private void detectGaps() {
-        // Skip if already running
-        if (isAutonomousBackfillRunning()) {
-            LOGGER.log(TRACE, "Gap detection already in progress, skipping this execution");
+    private void greedyBackfillRecentBlocks(long lastAcknowledgedBlockObserved, long lastAcknowledgedBlock) {
+        if (!backfillConfiguration.greedy()) {
             return;
         }
 
-        // Skip if OnDemand backfill is running
-        if (isOnDemandBackfillRunning()) {
-            LOGGER.log(TRACE, "On-Demand backfill is running, skipping autonomous gap detection");
+        // Skip if lack acknowledged block observed has increased
+        if (lastAcknowledgedBlockObserved > 0 && lastAcknowledgedBlock > lastAcknowledgedBlockObserved) {
+            LOGGER.log(
+                    TRACE,
+                    "Last acknowledged block observed has increased from {0} to {1}, skipping greedy backfill",
+                    lastAcknowledgedBlockObserved,
+                    lastAcknowledgedBlock);
             return;
         }
 
         try {
-            LOGGER.log(TRACE, "Detecting gaps in historical blocks");
-
-            // Calculate total missing blocks
-            long pendingBlocks = 0;
-
-            // Check for gaps between ranges
-            List<LongRange> blockRanges = context.historicalBlockProvider()
-                    .availableBlocks()
-                    .streamRanges()
-                    .toList();
+            LOGGER.log(TRACE, "Greedy backfilling recent blocks to stay close to network");
 
             // greedy backfill newer blocks available from peer BN sources
             detectedGaps = new ArrayList<>();
-            LongRange detectedRecentGapRange = backfillGrpcClientAutonomous.getNewAvailableRange(
-                    blockRanges.getLast().end());
+            LongRange detectedRecentGapRange = backfillGrpcClientAutonomous.getNewAvailableRange(lastAcknowledgedBlock);
+
+            // to-do: consider on-demand backfill range and remove from detectedRecentGapRange if overlapping
+            if (isOnDemandBackfillRunning()) {
+                // if on-demand backfill is running, we need to adjust the detected recent gap range
+            }
+
             if (detectedRecentGapRange.size() > 0 && detectedRecentGapRange.start() >= 0) {
                 detectedGaps.add(detectedRecentGapRange);
-                pendingBlocks += detectedRecentGapRange.size();
 
                 // backfill recent gaps first to prioritize staying close to the network
                 LOGGER.log(
                         TRACE,
                         "Detected recent gaps, numGaps={0} totalMissingBlocks={1}",
                         detectedGaps.size(),
-                        pendingBlocks);
+                        detectedRecentGapRange.size());
 
                 autonomousBackfillEndBlock.set(detectedRecentGapRange.end());
                 processDetectedGaps();
@@ -304,6 +308,51 @@ private void detectGaps() {
                 LOGGER.log(TRACE, "No recent gaps detected from other block node sources");
             }
 
+            autonomousError = false;
+        } catch (Exception e) {
+            LOGGER.log(TRACE, "Error during backfill autonomous process: {0}", e);
+            autonomousError = true;
+            autonomousBackfillEndBlock.set(-1);
+        }
+    }
+
+    private void detectGaps() {
+        // Skip if already running
+        if (isAutonomousBackfillRunning()) {
+            LOGGER.log(
+                    TRACE,
+                    "Autonomous backfill is already running up to {0}, skipping autonomous gap detection",
+                    autonomousBackfillEndBlock.get());
+            return;
+        }
+
+        // Skip if OnDemand backfill is running
+        if (isOnDemandBackfillRunning() && !backfillConfiguration.greedy()) {
+            LOGGER.log(
+                    TRACE,
+                    "On-Demand backfill is running starting from block {0}, skipping autonomous gap detection",
+                    onDemandBackfillStartBlock.get());
+            return;
+        }
+
+        try {
+            LOGGER.log(TRACE, "Detecting gaps in blocks");
+
+            // Calculate total missing blocks
+            long pendingBlocks = 0;
+
+            // Check for gaps between ranges
+            List<LongRange> blockRanges = context.historicalBlockProvider()
+                    .availableBlocks()
+                    .streamRanges()
+                    .toList();
+
+            // greedy backfill newer blocks available from peer BN sources
+            greedyBackfillRecentBlocks(
+                    lastAcknowledgedBlockObserved.get(), blockRanges.getLast().end());
+            lastAcknowledgedBlockObserved.set(
+                    blockRanges.getLast().end()); // update the last observed acknowledged block
+
             // backfill missing historical blocks from peer BN sources
             detectedGaps = new ArrayList<>();
             long expectedFirstBlock = backfillConfiguration.startBlock();
@@ -512,24 +561,24 @@ public void handleNewestBlockKnownToNetwork(NewestBlockKnownToNetworkNotificatio
             return;
         }
 
-        // we should create new Gap and a new task to backfill it
+        // we should create  new Gap and a new task to backfill it
         long lastPersistedBlock =
                 context.historicalBlockProvider().availableBlocks().max();
         long newestBlockKnown = notification.blockNumber();
-        final LongRange gap;
-        if (newestBlockKnown == UNKNOWN_BLOCK_NUMBER) {
-            // proactively get the new available range from other BN sources
-            gap = backfillGrpcClientOnDemand.getNewAvailableRange(lastPersistedBlock);
-        } else {
-            // create gap from last persisted + 1 to block number from notification
-            gap = new LongRange(lastPersistedBlock + 1, newestBlockKnown);
-        }
+        LongRange gap = new LongRange(lastPersistedBlock + 1, newestBlockKnown);
+        LOGGER.log(
+                TRACE,
+                "Detected new block known to network: {0,number,#}, starting backfill task for gap: {1}",
+                newestBlockKnown,
+                gap);
+
+        lastAcknowledgedBlockObserved.set(lastPersistedBlock); // update the last observed acknowledged block
 
         // if the gap is not empty, we can backfill it
         if (gap.size() > 0) {
             try {
-                // Skip if autonomous backfill is running
-                if (isAutonomousBackfillRunning()) {
+                // Skip if greedy autonomous backfill is greedy is more aggressive and will like cover more blocks
+                if (isAutonomousBackfillRunning() && backfillConfiguration.greedy()) {
                     LOGGER.log(TRACE, "Autonomous backfill is running, skipping on-demand gap detection");
                     return;
                 }
@@ -543,6 +592,9 @@ public void handleNewestBlockKnownToNetwork(NewestBlockKnownToNetworkNotificatio
                     try {
                         LOGGER.log(TRACE, "Starting on-demand backfill for gap: {0}", gap);
                         backfillGap(gap, BackfillType.ON_DEMAND);
+                        lastAcknowledgedBlockObserved.set(context.historicalBlockProvider()
+                                .availableBlocks()
+                                .max()); // update the last observed acknowledged block
                     } catch (ParseException | InterruptedException | RuntimeException e) {
                         LOGGER.log(TRACE, "Error backfilling gap {0}: {1}", gap, e);
                         backfillFetchErrors.add(1);