Various address assignment optimizations (#223)

* Add context to error logs
* Implement block depth threading
* Log thread work status
* gitignore cache files
* Add --min-block-depth flag
   * Allows user to set the minimum block depth of an address as used to calculate the work when distributing addresses to threads.
* Alternate over/under allocating work to threads
This ensures the last thread is closer to being evenly allocated compared to all the prior threads.
* Add --split-sync-threads option
    * This options splits up threads into synced and unsynced classes
* Improve logging of thread assignments
* Add support for balance-new-addresses
  * Option will more intelligently distribute new addresses to threads with a scan height that closely matches that of the incoming address
* fixup! Add support for balance-new-addresses
* Fix scanner_options and program objects sort order

---------

Co-authored-by: William Swanson <swansontec@gmail.com>

.gitignore

@@ -1 +1,2 @@
 /build
+/.cache/

docs/balance_new_addresses.md

@@ -0,0 +1,122 @@
+# Balance New Addresses
+
+## Overview
+
+The `--balance-new-addresses` option changes how new addresses are assigned to scanner threads while the scanner is actively running. Instead of using round-robin distribution, addresses are assigned to threads based on their current scanning progress, ensuring better workload distribution and faster synchronization.
+
+## Motivation
+
+The default round-robin algorithm assigns new addresses sequentially to threads without considering their current state. This can lead to:
+- New addresses being assigned to threads that are far behind, causing unnecessary delays
+- Imbalanced workload when addresses with different scan heights are added
+- Inefficient thread utilization when threads finish at different times
+
+The balance-new-addresses algorithm addresses these issues by intelligently matching new addresses to threads based on their scanning progress.
+
+## Configuration
+
+- `--balance-new-addresses`: Enable balanced assignment of new addresses to threads (default: false)
+
+## Usage
+
+Enable balanced new address assignment:
+```bash
+monero-lws-daemon --balance-new-addresses [other options]
+```
+
+Or in config file:
+```
+balance-new-addresses=true
+```
+
+## Algorithm
+
+### Thread Selection Criteria
+
+When a new address is added (e.g., an inactive account becomes active), the algorithm selects a thread using the following priority:
+
+1. **Primary preference**: Thread with the highest scan height that is **not above** the account's scan height
+   - If multiple threads meet this criteria, choose the one with the fewest addresses
+   
+2. **Fallback**: If all threads are above the account's scan height, choose the thread with the **lowest** scan height
+   - This minimizes backtracking when all threads have already passed the account's starting point
+   - If multiple threads have the same lowest height, choose the one with the fewest addresses
+
+### Scan Height Tracking
+
+Each thread maintains a `current_min_height` value representing the minimum scan height of all accounts currently assigned to that thread. This value is:
+- Updated when new accounts are added to the thread (if the new account has a lower scan height)
+- Updated periodically as the thread processes blocks and advances its scanning progress
+- Used by the algorithm to determine which thread is best suited for a new account
+
+### Scope
+
+- **Only affects local threads**: The algorithm only applies to threads running on the local daemon. When enabled and local threads are available, **all new addresses are assigned exclusively to local threads**. Remote scanner clients will not receive any new addresses unless there are zero local threads available.
+- **Only affects new addresses**: This algorithm is used when addresses are added dynamically (e.g., inactive accounts becoming active). Initial thread assignment at startup uses the standard block-depth-threading or round-robin algorithms.
+- **Works with other threading options**: Can be used alongside `--block-depth-threading` and `--split-synced` options.
+
+## Example
+
+Consider a scenario with 3 local threads and a new account that needs to be assigned:
+
+**Thread states:**
+- Thread 0: scanning at height 3,000,000 (5 accounts)
+- Thread 1: scanning at height 3,100,000 (3 accounts)
+- Thread 2: scanning at height 2,900,000 (7 accounts)
+
+**Scenario 1: New account at height 3,050,000**
+- Thread 0: height 3,000,000 ≤ 3,050,000 ✓
+- Thread 1: height 3,100,000 > 3,050,000 ✗
+- Thread 2: height 2,900,000 ≤ 3,050,000 ✓
+- **Result**: Thread 0 is selected (highest height ≤ account height, and fewer accounts than Thread 2)
+
+**Scenario 2: New account at height 2,800,000 (all threads are above)**
+- Thread 0: height 3,000,000 > 2,800,000 ✗
+- Thread 1: height 3,100,000 > 2,800,000 ✗
+- Thread 2: height 2,900,000 > 2,800,000 ✗
+- **Result**: Thread 2 is selected (lowest height among all threads, minimizing backtracking)
+
+**Scenario 3: New account at height 3,200,000 (all threads are below)**
+- Thread 0: height 3,000,000 ≤ 3,200,000 ✓
+- Thread 1: height 3,100,000 ≤ 3,200,000 ✓
+- Thread 2: height 2,900,000 ≤ 3,200,000 ✓
+- **Result**: Thread 1 is selected (highest height ≤ account height, and fewer accounts than Thread 0)
+
+## Comparison with Round-Robin
+
+**Round-Robin Algorithm** (default):
+- Assigns addresses sequentially: Thread 0, Thread 1, Thread 2, Thread 0, ...
+- Does not consider thread state or account scan height
+- Simple and predictable, but can lead to suboptimal assignments
+
+**Balance New Addresses Algorithm**:
+- Considers each thread's current scanning progress
+- Matches accounts to threads based on scan height compatibility
+- More complex, but provides better workload distribution and faster synchronization
+
+## When New Addresses Are Added
+
+The balance-new-addresses algorithm is triggered when:
+- An inactive account becomes active (status changes from inactive to active)
+- New accounts are detected during periodic checks (every 10 seconds)
+
+It is **not** triggered by:
+- Initial thread assignment at startup (uses block-depth-threading or round-robin)
+- Full account reassignment (e.g., after rescan operations)
+- Accounts assigned to remote scanner clients
+
+## Benefits
+
+- **Faster synchronization**: New addresses are assigned to threads that are already at or near their scan height
+- **Reduced backtracking**: Minimizes cases where threads need to scan backwards to process new accounts
+- **Better workload distribution**: Accounts are distributed based on actual thread progress rather than arbitrary rotation
+- **Improved thread utilization**: Threads that are ahead receive new work more efficiently
+
+## Limitations
+
+- **Remote threads excluded**: When enabled and local threads are available, remote scanner clients will not receive any new addresses. Remote threads only receive work if there are zero local threads available.
+- Does not affect initial thread assignment at startup
+- Requires tracking thread scan heights, which adds minimal overhead
+- May not provide benefits if all threads are at similar heights
+- **Important**: If you have remote scanner clients and want them to receive new addresses, do not enable this option, or ensure you have zero local scanner threads configured
+

docs/block_depth_threading.md

@@ -0,0 +1,106 @@
+# Block Depth Threading
+
+## Overview
+
+Block depth threading is a work distribution algorithm that balances scanner thread workload based on the amount of blockchain data each address needs to process, rather than simply distributing addresses evenly across threads. This approach addresses the performance inefficiency where threads with fewer blocks to scan finish early and become idle while other threads continue processing.
+
+## Motivation
+
+The default threading algorithm distributes addresses evenly across threads, treating each address as an equal unit of work. However, addresses have vastly different synchronization requirements:
+- A fully synced address may only need to process a few recent blocks
+- A newly added address may need to scan hundreds of thousands of blocks
+
+This imbalance causes threads with mostly synced addresses to finish quickly and remain idle while threads with unsynced addresses continue working, resulting in poor CPU utilization and longer overall sync times.
+
+## Configuration
+
+- `--block-depth-threading`: Enable block depth threading algorithm (default: false)
+- `--min-block-depth`: Minimum block depth value for workload calculations (default: 16)
+
+## Usage
+
+Enable block depth threading with default settings:
+```bash
+monero-lws-daemon --block-depth-threading [other options]
+```
+
+Enable with custom minimum block depth:
+```bash
+monero-lws-daemon --block-depth-threading --min-block-depth=32 [other options]
+```
+
+Or in config file:
+```
+block-depth-threading=true
+min-block-depth=32
+```
+
+## Algorithm
+
+### Block Depth Calculation
+
+For each address, the **block depth** is calculated as the number of blocks remaining to be scanned:
+
+```
+blockdepth = max(current_blockchain_height - address_scan_height, min_block_depth)
+```
+
+Where `min_block_depth` is the value specified by `--min-block-depth` (default: 16).
+
+Addresses with blockdepth less than the minimum are assigned the minimum value. This prevents edge cases where fully synced addresses would have zero blockdepth, which could cause:
+- Division by zero or near-zero values in workload calculations
+- Degenerate cases where many fully-synced accounts get assigned together
+- Poor workload distribution when most accounts are fully synced
+
+### Minimum Block Depth
+
+The `--min-block-depth` flag sets the minimum block depth value used in workload calculations. This ensures that even fully synced accounts contribute meaningfully to workload balancing.
+
+**Default value**: 16 blocks
+
+**Example**: With `--min-block-depth=16`, an account at the current blockchain height (0 blocks remaining) is treated as having 16 blocks remaining for workload distribution purposes.
+
+**When to adjust**: You may want to increase this value if you have many fully synced accounts and want them to contribute more to workload balancing, or decrease it if you want more precise distribution for nearly-synced accounts.
+
+### Thread Assignment
+
+1. **Calculate total work**: Sum all address blockdepths to get `total_blockdepth`
+2. **Calculate target per thread**: `blockdepth_per_thread = total_blockdepth / thread_count`
+3. **Sort addresses**: Order by blockdepth (smallest first)
+4. **Distribute to threads**: 
+   - Addresses are assigned sequentially to threads
+   - Accounts are added to the current thread until the cumulative depth reaches or exceeds the target
+   - When target is reached, move to the next thread
+   - Final thread receives any remaining addresses
+
+This overallocation strategy ensures more balanced workload distribution and better thread utilization throughout the scanning process.
+
+## Example
+
+With 4 threads and 20 accounts with varying sync states:
+- Accounts A-H: 16 blocks each (synced, at minimum) = 128 blocks
+- Accounts I-L: 100 blocks each = 400 blocks
+- Accounts M-P: 300 blocks each = 1,200 blocks
+- Accounts Q-T: 500 blocks each = 2,000 blocks
+
+**Old Algorithm** (by count, evenly distributed - 5 accounts per thread):
+- Thread 0: A, B, C, D, E (80 blocks) ✓ finishes immediately
+- Thread 1: F, G, H, I, J (228 blocks) ⏱
+- Thread 2: K, L, M, N, O (1,028 blocks) ⏱⏱
+- Thread 3: P, Q, R, S, T (2,100 blocks) ⏱⏱⏱⏱ takes much longer
+- **Problem**: Despite equal account count (5 per thread), massive workload imbalance - thread 3 has 26x more work than thread 0
+
+**New Algorithm** (by depth, balanced workload with alternating over/under allocation):
+- Total: 3,728 blocks, target: 932 blocks/thread
+- Thread 0: (even, over-allocate): A, B, C, D, E, F, G, H, I, J, K, L, M, N (1,128 blocks)
+- Thread 1: (odd, under-allocate): O, P (600 blocks)
+- Thread 2: (even, over-allocate): Q, R (1,000 blocks)
+- Thread 3: (odd, under-allocate): S, T (1,000 blocks)
+- **Result**: All 4 threads utilized with better balance (600-1,128 vs 80-2,100 blocks), synced accounts efficiently grouped
+
+## Benefits
+
+- **Improved parallelization**: All threads remain active longer
+- **Reduced sync time**: More efficient CPU utilization
+- **Better resource usage**: Eliminates idle threads waiting for others to complete
+- **Predictable performance**: Workload is distributed based on actual work required

docs/split_synced_threads.md

@@ -0,0 +1,84 @@
+# Split Synced Threading
+
+## Overview
+
+The split-sync threading feature is an enhancement to the block-depth-threading algorithm that isolates synced addresses from unsynced addresses across different scanner threads. This prevents fully or near-fully synced addresses from being blocked by addresses that are still catching up to the blockchain.
+
+Split-sync is enabled by setting `--split-sync-threads` to a value greater than 0, which specifies the percentage of threads to allocate for synced accounts.
+
+## Requirements
+
+- Requires `--block-depth-threading`
+- Only affects initial thread assignment at startup
+- `--split-sync-threads` accepts a value from 0-1 representing the percentage of threads for synced accounts (e.g., 0.25 = 25%)
+- `--split-sync-depth` accepts a numeric value representing the maximum block depth for an address to be considered synced (defaults to 10)
+- If `--split-sync-threads` is not provided or set to 0, split-sync is disabled
+
+## Arguments
+
+- `--split-sync-threads=[numthreadspercent]`: Specified as a number from 0-1, where 0.25 = 25%. Defaults to 0 (disabled) if not specified. Split-sync is enabled when this value is > 0.
+- `--split-sync-depth=[synced depth]`: The maximum block depth for an account to be considered synced. Defaults to 10 if not specified. Accounts with `raw_blockdepth ≤ split-sync-depth` are considered synced.
+
+## Behavior
+
+### Address Classification
+
+Addresses are classified as either **synced** or **unsynced** based on their block depth:
+- **Synced address**: `raw_blockdepth ≤ --split-sync-depth` value (default 10)
+- **Unsynced address**: `raw_blockdepth > --split-sync-depth` value
+
+Note: The `--min-block-depth` parameter is only used for assigning a minimum block depth value to addresses for workload calculations, and is not used to determine if an address is synced.
+
+### Thread Allocation Algorithm
+
+When `--split-sync-threads > 0`:
+
+1. **Calculate synced thread count**: The number of threads allocated for synced accounts is calculated as `ceil(split-sync-threads × total_threads)`, rounded up. The remaining threads are used for unsynced accounts.
+
+2. **Separate accounts**: Accounts are classified as synced or unsynced based on `--split-sync-depth`.
+
+3. **Distribute synced accounts**: Synced accounts are distributed across the allocated synced threads using **round-robin assignment** (account 0 → thread 0, account 1 → thread 1, ..., account N → thread (N % num_synced_threads)). Accounts are already sorted by block depth (smallest to largest) from the block-depth-threading preparation.
+
+4. **Distribute unsynced accounts**: Unsynced accounts are distributed across the remaining threads using the **standard block-depth-threading algorithm** (alternating over/under-allocation strategy) starting from the first unsynced thread.
+
+### Edge Cases
+
+- **No synced accounts**: If there are no synced accounts, all threads are used for unsynced accounts with standard block-depth-threading.
+- **Fewer synced accounts than threads**: If there are fewer synced accounts than threads allocated for synced accounts, only as many synced threads are used as there are synced accounts (one account per thread). The remaining threads are used for unsynced accounts.
+- **Split-sync disabled**: If `--split-sync-threads` is 0 or not set (default), split-sync is disabled and all accounts use standard block-depth-threading.
+
+## Example
+
+With 4 threads, `--split-sync-threads=0.25`, and `--split-sync-depth=16`, and 20 accounts with varying sync states:
+- Accounts A-H: 16 blocks each (synced) = 128 blocks
+- Accounts I-L: 100 blocks each (unsynced) = 400 blocks
+- Accounts M-P: 300 blocks each (unsynced) = 1,200 blocks
+- Accounts Q-T: 500 blocks each (unsynced) = 2,000 blocks
+
+**Thread allocation**:
+- Synced threads: `ceil(0.25 × 4) = 1` thread
+- Unsynced threads: `4 - 1 = 3` threads
+
+**With split-sync enabled**:
+- Thread 0 (synced, round-robin): A, B, C, D, E, F, G, H (8 accounts)
+  - Contains: 8 synced (A-H) only - **synced thread**
+- Thread 1 (unsynced, block-depth-threading): I, J, K, L, M (800 blocks)
+  - Contains: 5 unsynced (I-M) only - **unsynced thread**
+- Thread 2 (unsynced, block-depth-threading): N, O, P, Q (1,400 blocks)
+  - Contains: 4 unsynced - **unsynced thread**
+- Thread 3 (unsynced, block-depth-threading): R, S, T (1,500 blocks)
+  - Contains: 3 unsynced - **unsynced thread**
+- **Result**: Synced addresses (A-H) are isolated in Thread 0 and can process updates quickly without waiting for unsynced addresses. All unsynced addresses are separated into threads 1-3 using block-depth-threading.
+
+**Without split-sync** (standard block-depth-threading):
+- Total: 3,728 blocks, target: 932 blocks/thread
+- Thread 0 (even, over-allocate): A, B, C, D, E, F, G, H, I, J, K, L, M, N (1,128 blocks)
+  - Contains: 8 synced (A-H) + 6 unsynced (I-N) - **mixed**
+- Thread 1 (odd, under-allocate): O, P (600 blocks)
+- Thread 2 (even, over-allocate): Q, R (1,000 blocks)
+- Thread 3 (odd, under-allocate): S, T (1,000 blocks)
+- **Problem**: Synced addresses (A-H) are mixed with unsynced addresses (I-N) in Thread 0, causing synced addresses to wait for unsynced ones to catch up
+
+## Benefits
+
+This approach ensures that synced addresses receive timely updates without being delayed by the potentially lengthy synchronization process of newly added or far-behind addresses. By pre-allocating threads and using round-robin distribution for synced accounts, the algorithm provides predictable thread allocation and better isolation between synced and unsynced workloads.

src/client_main.cpp

@@ -246,7 +246,7 @@ namespace
 
         if (!users.empty())
         {
-          static constexpr const lws::scanner_options opts{false, false, false};
+          static constexpr const lws::scanner_options opts{0, 0, lws::MINIMUM_BLOCK_DEPTH, 0, false, false, false, false};
 
           auto new_client = MONERO_UNWRAP(zclient.clone());
           MONERO_UNWRAP(new_client.watch_scan_signals());
@@ -259,7 +259,7 @@ namespace
     catch (const std::exception& e)
     {
       self.shutdown();
-      MERROR(e.what());
+      MERROR("Client shutdown with error " << e.what());
     }
     catch (...)
     {
@@ -375,7 +375,7 @@ int main(int argc, char** argv)
   }
   catch (std::exception const& e)
   {
-    MERROR(e.what());
+    MERROR("Client shutdown with error " << e.what());
     return EXIT_FAILURE;
   }
   catch (...)

src/rpc/scanner/queue.cpp

@@ -25,8 +25,10 @@
 // STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
 // THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
+#include "db/data.h"
 #include "queue.h"
 
+#include <algorithm>  // std::min_element
 #include "db/account.h"
 
 namespace lws { namespace rpc { namespace scanner 
@@ -43,7 +45,7 @@ namespace lws { namespace rpc { namespace scanner
   }
 
   queue::queue()
-    : replace_(), push_(), user_count_(0), sync_(), poll_(), stop_(false)
+    : replace_(), push_(), user_count_(0), current_min_height_(db::block_id(0)), sync_(), poll_(), stop_(false)
   {}
 
   queue::~queue()
@@ -64,6 +66,18 @@ namespace lws { namespace rpc { namespace scanner
     return user_count_;
   }
 
+  db::block_id queue::current_min_height()
+  {
+    const boost::lock_guard<boost::mutex> lock{sync_};
+    return current_min_height_;
+  }
+
+  void queue::update_min_height(db::block_id height)
+  {
+    const boost::lock_guard<boost::mutex> lock{sync_};
+    current_min_height_ = std::max(height, current_min_height_);
+  }
+
   queue::status queue::get_accounts()
   {
     const boost::lock_guard<boost::mutex> lock{sync_};
@@ -85,6 +99,20 @@ namespace lws { namespace rpc { namespace scanner
       replace_ = std::move(users);
       user_count_ = replace_->size();
       push_.clear();
+      // Update min height based on the oldest account in the replacement set
+      if (!replace_->empty())
+      {
+        current_min_height_ = std::min_element(
+          replace_->begin(), replace_->end(),
+          [](const lws::account& a, const lws::account& b) {
+            return a.scan_height() < b.scan_height();
+          }
+        )->scan_height();
+      }
+      else
+      {
+        current_min_height_ = db::block_id(0);
+      }
     }
     poll_.notify_all();
   }

src/rpc/scanner/queue.h

@@ -26,6 +26,7 @@
 // THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 #pragma once
 
+#include <algorithm>
 #include <boost/thread/condition_variable.hpp>
 #include <boost/thread/lock_guard.hpp>
 #include <boost/thread/mutex.hpp>
@@ -54,6 +55,7 @@ namespace lws { namespace rpc { namespace scanner
     std::optional<std::vector<lws::account>> replace_;
     std::vector<lws::account> push_;
     std::size_t user_count_;
+    db::block_id current_min_height_;  //!< Minimum scan height of accounts on this thread
     boost::mutex sync_;
     boost::condition_variable poll_;
     bool stop_;
@@ -70,6 +72,12 @@ namespace lws { namespace rpc { namespace scanner
     //! \return Total number of users given to this local thread
     std::size_t user_count();
 
+    //! \return Current minimum scan height of accounts on this thread
+    db::block_id current_min_height();
+
+    //! Update the current minimum scan height for this thread
+    void update_min_height(db::block_id height);
+
     //! \return Replacement and "push" accounts
     status get_accounts();
 
@@ -86,6 +94,12 @@ namespace lws { namespace rpc { namespace scanner
       {
         const boost::lock_guard<boost::mutex> lock{sync_};
         user_count_ += (end - begin);
+        // Update min height if any new account has a lower scan height
+        for (auto it = begin; it != end; ++it)
+        {
+          if (current_min_height_ == db::block_id(0) || it->scan_height() < current_min_height_)
+            current_min_height_ = it->scan_height();
+        }
         push_.insert(push_.end(), std::make_move_iterator(begin), std::make_move_iterator(end));
       }
       poll_.notify_all();

src/rpc/scanner/server.cpp

@@ -32,6 +32,7 @@
 #include <boost/asio/dispatch.hpp>
 #include <boost/lexical_cast.hpp>
 #include <boost/numeric/conversion/cast.hpp>
+#include <limits>
 #include <sodium/utils.h>
 #include <sodium/randombytes.h>
 #include <vector>
@@ -287,39 +288,97 @@ namespace lws { namespace rpc { namespace scanner
         return;
       }
 
-      self_->next_thread_ %= total_threads;
+      if (self_->balance_new_addresses_ && !self_->local_.empty())
-      while (!new_accounts.empty())
       {
-        if (self_->next_thread_ < self_->local_.size())
+        // New algorithm: Assign new accounts to thread with highest blockheight not above account height
+        // Only supports local scanning for now since remote clients can't share their scan heights.
+        while (!new_accounts.empty())
         {
-          self_->local_[self_->next_thread_]->push_accounts(
+          const db::block_id account_height = new_accounts.back().scan_height();
+          std::size_t best_thread = 0;
+          db::block_id best_height = db::block_id(0);
+          std::size_t best_count = std::numeric_limits<std::size_t>::max();
+          bool found_below_or_equal = false;
+          
+          // Check local threads
+          for (std::size_t i = 0; i < self_->local_.size(); ++i)
+          {
+            const db::block_id height = self_->local_[i]->current_min_height();
+            const std::size_t count = self_->local_[i]->user_count();
+            
+            if (height <= account_height)
+            {
+              // Prefer thread with highest height <= account_height, or if tied, fewer addresses
+              if (!found_below_or_equal || height > best_height || (height == best_height && count < best_count))
+              {
+                found_below_or_equal = true;
+                best_thread = i;
+                best_height = height;
+                best_count = count;
+              }
+            }
+            else
+            {
+              // Thread is above account height - only consider if no thread below/equal found yet
+              if (!found_below_or_equal)
+              {
+                // Prefer thread with lowest height, or if tied, fewer addresses
+                if (best_height == db::block_id(0) || height < best_height || (height == best_height && count < best_count))
+                {
+                  best_thread = i;
+                  best_height = height;
+                  best_count = count;
+                }
+              }
+            }
+          }
+          
+          MINFO("Thread " << best_thread << " at height " << std::uint64_t(best_height) 
+                << " received new account at height " << std::uint64_t(account_height));
+          self_->local_[best_thread]->push_accounts(
             std::make_move_iterator(new_accounts.end() - 1),
             std::make_move_iterator(new_accounts.end())
           );
           new_accounts.erase(new_accounts.end() - 1);
-          ++self_->next_thread_;
         }
-        else
+      }
+      else
+      {
+        // Original round-robin algorithm
+        self_->next_thread_ %= total_threads;
+        while (!new_accounts.empty())
         {
-          std::size_t j = 0;
+          if (self_->next_thread_ < self_->local_.size())
-          for (auto offset = self_->local_.size(); j < remotes.size(); ++j)
           {
-            if (self_->next_thread_ <= offset)
+            self_->local_[self_->next_thread_]->push_accounts(
-              break;
+              std::make_move_iterator(new_accounts.end() - 1),
-            offset += remotes[j]->threads_;
+              std::make_move_iterator(new_accounts.end())
+            );
+            new_accounts.erase(new_accounts.end() - 1);
+            ++self_->next_thread_;
+          }
+          else
+          {
+            std::size_t j = 0;
+            for (auto offset = self_->local_.size(); j < remotes.size(); ++j)
+            {
+              if (self_->next_thread_ <= offset)
+                break;
+              offset += remotes[j]->threads_;
+            }
+
+            const auto user_count = std::min(new_accounts.size(), remotes[j]->threads_);
+            std::vector<lws::account> next{
+              std::make_move_iterator(new_accounts.end() - user_count),
+              std::make_move_iterator(new_accounts.end())
+            };
+            new_accounts.erase(new_accounts.end() - user_count);
+            write_command(remotes[j], push_accounts{std::move(next)});
+            self_->next_thread_ += remotes[j]->threads_;
           }
 
-          const auto user_count = std::min(new_accounts.size(), remotes[j]->threads_);
+          self_->next_thread_ %= total_threads;
-          std::vector<lws::account> next{
-            std::make_move_iterator(new_accounts.end() - user_count),
-            std::make_move_iterator(new_accounts.end())
-          };
-          new_accounts.erase(new_accounts.end() - user_count);
-          write_command(remotes[j], push_accounts{std::move(next)});
-          self_->next_thread_ += remotes[j]->threads_;
         }
-
-        self_->next_thread_ %= total_threads;
       }
       reader->finish_read();
       self_->accounts_cur_ = current_users.give_cursor();
@@ -460,7 +519,7 @@ namespace lws { namespace rpc { namespace scanner
     };
   }
 
-  server::server(boost::asio::io_context& io, db::storage disk, rpc::client zclient, std::vector<std::shared_ptr<queue>> local, std::vector<db::account_id> active, std::shared_ptr<boost::asio::ssl::context> ssl)
+  server::server(boost::asio::io_context& io, db::storage disk, rpc::client zclient, std::vector<std::shared_ptr<queue>> local, std::vector<db::account_id> active, std::shared_ptr<boost::asio::ssl::context> ssl, bool balance_new_addresses)
     : strand_(io),
       check_timer_(io),
       acceptor_(io),
@@ -474,7 +533,8 @@ namespace lws { namespace rpc { namespace scanner
       next_thread_(0),
       pass_hashed_(),
       pass_salt_(),
-      stop_(false)
+      stop_(false),
+      balance_new_addresses_(balance_new_addresses)
   {
     std::sort(active_.begin(), active_.end());
     for (const auto& local : local_)

src/rpc/scanner/server.h

@@ -72,6 +72,7 @@ namespace lws { namespace rpc { namespace scanner
     std::array<unsigned char, 32> pass_hashed_;
     std::array<unsigned char, crypto_pwhash_SALTBYTES> pass_salt_;
     bool stop_;
+    bool balance_new_addresses_;
 
     //! Async acceptor routine
     class acceptor;
@@ -86,7 +87,7 @@ namespace lws { namespace rpc { namespace scanner
   public:
     static boost::asio::ip::tcp::endpoint get_endpoint(const std::string& address);
 
-    explicit server(boost::asio::io_context& io, db::storage disk, rpc::client zclient, std::vector<std::shared_ptr<queue>> local, std::vector<db::account_id> active, std::shared_ptr<boost::asio::ssl::context> ssl);
+    explicit server(boost::asio::io_context& io, db::storage disk, rpc::client zclient, std::vector<std::shared_ptr<queue>> local, std::vector<db::account_id> active, std::shared_ptr<boost::asio::ssl::context> ssl, bool balance_new_addresses = false);
 
     server(const server&) = delete;
     server(server&&) = delete;

src/scanner.cpp

@@ -34,6 +34,7 @@
 #include <boost/thread/mutex.hpp>
 #include <boost/thread/thread.hpp>
 #include <cassert>
+#include <cmath>
 #include <chrono>
 #include <cstring>
 #include <functional>
@@ -651,20 +652,30 @@ namespace lws
           }
 
           users.clear();
+          MINFO("Thread " << thread_n << " has gone idle, waiting for work");
           auto status = queue->wait_for_accounts();
           if (status.replace)
+          {
             users = std::move(*status.replace);
-          users.insert(
+            MINFO("Thread " << thread_n << " received " << users.size() << " replacement account(s), starting work");
-            users.end(),
+          }
-            std::make_move_iterator(status.push.begin()),
+          if (!status.push.empty())
-            std::make_move_iterator(status.push.end())
+          {
-          );
+            MINFO("Thread " << thread_n << " received " << status.push.size() << " new account(s)");
+            users.insert(
+              users.end(),
+              std::make_move_iterator(status.push.begin()),
+              std::make_move_iterator(status.push.end())
+            );
+            if (!status.replace)
+              MINFO("Thread " << thread_n << " now has " << users.size() << " total account(s), starting work");
+          }
         }
       }
       catch (std::exception const& e)
       {
         self.shutdown();
-        MERROR(e.what());
+        MERROR("Scanner shutdown with error " << e.what());
       }
       catch (...)
       {
@@ -975,6 +986,9 @@ namespace lws
 
           for (account& user : users)
             user.updated(db::block_id(fetched->start_height));
+          
+          // Update queue with current minimum scan height (oldest account determines thread progress)
+          queue.update_min_height(users.front().scan_height());
         }
       }
 
@@ -1044,25 +1058,230 @@ namespace lws
         boost::thread::attributes attrs;
         attrs.set_stack_size(THREAD_STACK_SIZE);
 
-        std::sort(users.begin(), users.end(), by_height{});
-
         MINFO("Starting scan loops on " << thread_count << " thread(s) with " << users.size() << " account(s)");
 
-        for (std::size_t i = 0; i < queues.size(); ++i)
+        if (opts.block_depth_threading)
         {
-          queues[i] = std::make_shared<rpc::scanner::queue>();
+          // Get current blockchain height
-
+          const db::block_id current_height = MONERO_UNWRAP(MONERO_UNWRAP(disk.start_read()).get_last_block()).id;
-          // this can create threads with no active accounts, they just wait
+          
-          const std::size_t count = users.size() / (queues.size() - i);
+          // Calculate blockdepth for each account and create pairs
-          std::vector<lws::account> thread_users{
+          struct account_depth {
-            std::make_move_iterator(users.end() - count), std::make_move_iterator(users.end())
+            std::size_t index;
+            std::uint64_t blockdepth;  // Adjusted blockdepth (with min_block_depth applied) for workload calculations
+            std::uint64_t raw_blockdepth;  // True blockdepth for split-sync classification
           };
-          users.erase(users.end() - count, users.end());
+          std::vector<account_depth> account_depths;
+          account_depths.reserve(users.size());
+          
+          std::uint64_t total_blockdepth = 0;
+          for (std::size_t i = 0; i < users.size(); ++i)
+          {
+            const std::uint64_t raw_blockdepth = std::uint64_t(current_height) - std::uint64_t(users[i].scan_height());
+            const std::uint64_t blockdepth = std::max(raw_blockdepth, opts.min_block_depth);
+            account_depths.push_back(account_depth{i, blockdepth, raw_blockdepth});
+            total_blockdepth += blockdepth;
+          }
+          
+          // Sort by raw_blockdepth (smallest first) to group accounts by true block depth
+          std::sort(account_depths.begin(), account_depths.end(),
+            [](const account_depth& a, const account_depth& b) {
+              return a.raw_blockdepth < b.raw_blockdepth;
+            });
+          
+          // Prepare thread assignment data structure
+          std::vector<std::vector<lws::account>> thread_assignments(thread_count);
+          
+          // Track min/max raw block depths for each thread (for logging true block depth ranges)
+          std::vector<std::uint64_t> thread_min_raw_depth(thread_count, std::numeric_limits<std::uint64_t>::max());
+          std::vector<std::uint64_t> thread_max_raw_depth(thread_count, 0);
+          
+          // Initialize for standard block-depth-threading (all accounts, all threads)
+          std::vector<account_depth> accounts_to_assign = account_depths;
+          std::size_t start_thread = 0;
+          std::size_t num_threads_for_assignment = thread_count;
+          
+          // If split-sync is enabled, separate and assign synced accounts first
+          if (opts.split_sync_threads > 0.0)
+          {
+            // Separate synced and unsynced accounts
+            std::vector<account_depth> synced_accounts;
+            std::vector<account_depth> unsynced_accounts;
+            synced_accounts.reserve(account_depths.size());
+            unsynced_accounts.reserve(account_depths.size());
+            
+            for (const auto& ad : account_depths)
+            {
+              if (ad.raw_blockdepth <= opts.split_sync_depth)
+                synced_accounts.push_back(ad);
+              else
+                unsynced_accounts.push_back(ad);
+            }
+            
+            // Handle case: no synced accounts - use all threads for unsynced
+            if (synced_accounts.empty())
+            {
+              MINFO("No synced accounts, using all threads for unsynced accounts");
+              accounts_to_assign = std::move(unsynced_accounts);
+            }
+            else
+            {
+              // Calculate number of threads for synced accounts (rounded up)
+              const std::size_t num_synced_threads = std::max(std::size_t(1), 
+                static_cast<std::size_t>(std::ceil(opts.split_sync_threads * thread_count)));
+              
+              // Distribute synced accounts using round-robin
+              // Use only as many threads as we have synced accounts (if fewer)
+              const std::size_t actual_synced_threads = std::min(num_synced_threads, synced_accounts.size());
+              
+              for (std::size_t i = 0; i < synced_accounts.size(); ++i)
+              {
+                const std::size_t thread_idx = i % actual_synced_threads;
+                const auto& ad = synced_accounts[i];
+                thread_assignments[thread_idx].push_back(std::move(users[ad.index]));
+                
+                // Update min/max raw block depths for current thread (for logging)
+                if (ad.raw_blockdepth < thread_min_raw_depth[thread_idx])
+                  thread_min_raw_depth[thread_idx] = ad.raw_blockdepth;
+                if (ad.raw_blockdepth > thread_max_raw_depth[thread_idx])
+                  thread_max_raw_depth[thread_idx] = ad.raw_blockdepth;
+              }
+              
+              // Remaining accounts (unsynced) will be assigned using block-depth-threading
+              accounts_to_assign = std::move(unsynced_accounts);
+              start_thread = num_synced_threads;
+              num_threads_for_assignment = thread_count - num_synced_threads;
+              
+              MINFO("Using split-sync threading: total_threads=" << thread_count
+                    << ", synced_threads=" << num_synced_threads
+                    << ", unsynced_threads=" << num_threads_for_assignment
+                    << ", split_sync_depth=" << opts.split_sync_depth
+                    << ", synced_accounts=" << synced_accounts.size()
+                    << ", unsynced_accounts=" << accounts_to_assign.size());
 
-          auto data = std::make_shared<thread_data>(
+              // Log the synced accounts assignments
-            MONERO_UNWRAP(ctx.connect()), disk.clone(), std::move(thread_users), queues[i], opts
+              for (std::size_t thread_idx = 0; thread_idx < actual_synced_threads; ++thread_idx)
-          );
+              {
-          threads.emplace_back(attrs, std::bind(&do_scan_loop, std::ref(self), std::move(data), i));
+                if (!thread_assignments[thread_idx].empty())
+                {
+                  const std::uint64_t min_depth = thread_min_raw_depth[thread_idx] == std::numeric_limits<std::uint64_t>::max() ? 0 : thread_min_raw_depth[thread_idx];
+                  const std::uint64_t max_depth = thread_max_raw_depth[thread_idx];
+                  MINFO("Thread " << thread_idx << " added " << thread_assignments[thread_idx].size() 
+                        << " addresses with true block depths " << min_depth << " to " << max_depth);
+                }
+              }
+            }
+          }
+          else
+          {
+            MINFO("Using block-depth threading: total_blockdepth=" << total_blockdepth 
+                  << ", min_block_depth=" << opts.min_block_depth);
+          }
+          
+          // Distribute remaining accounts using block-depth-threading
+          if (!accounts_to_assign.empty() && num_threads_for_assignment > 0)
+          {
+            // Calculate total blockdepth for accounts to assign
+            std::uint64_t assignment_total_blockdepth = 0;
+            for (const auto& ad : accounts_to_assign)
+              assignment_total_blockdepth += ad.blockdepth;
+            
+            const std::uint64_t blockdepth_per_thread = assignment_total_blockdepth / num_threads_for_assignment;
+            
+            // Assign accounts to threads based on cumulative blockdepth
+            std::size_t current_thread = start_thread;
+            std::uint64_t current_thread_depth = 0;
+            const std::size_t last_thread = start_thread + num_threads_for_assignment - 1;
+            
+            for (const auto& ad : accounts_to_assign)
+            {
+              // Alternate between over-allocating and under-allocating threads
+              // Even threads (0, 2, 4...): over-allocate - move to next when current depth reaches target
+              // Odd threads (1, 3, 5...): under-allocate - move to next when adding next account would exceed target
+              bool should_move_to_next_thread = false;
+              
+              if (current_thread < last_thread)
+              {
+                const std::size_t relative_thread = current_thread - start_thread;
+                if (relative_thread % 2 == 0)
+                {
+                  // Even thread: over-allocate (move when current depth >= target)
+                  should_move_to_next_thread = (current_thread_depth >= blockdepth_per_thread);
+                }
+                else
+                {
+                  // Odd thread: under-allocate (move when adding this account would exceed target)
+                  should_move_to_next_thread = (current_thread_depth + ad.blockdepth > blockdepth_per_thread);
+                }
+                
+                if (should_move_to_next_thread)
+                {
+                  // Log the current thread before moving to next
+                  if (!thread_assignments[current_thread].empty())
+                  {
+                    const std::uint64_t min_depth = thread_min_raw_depth[current_thread] == std::numeric_limits<std::uint64_t>::max() ? 0 : thread_min_raw_depth[current_thread];
+                    const std::uint64_t max_depth = thread_max_raw_depth[current_thread];
+                    MINFO("Thread " << current_thread << " added " << thread_assignments[current_thread].size() 
+                          << " addresses with true block depths " << min_depth << " to " << max_depth);
+                  }
+                  
+                  ++current_thread;
+                  current_thread_depth = 0;
+                }
+              }
+              
+              thread_assignments[current_thread].push_back(std::move(users[ad.index]));
+              current_thread_depth += ad.blockdepth;
+              
+              // Update min/max raw block depths for current thread (for logging)
+              if (ad.raw_blockdepth < thread_min_raw_depth[current_thread])
+                thread_min_raw_depth[current_thread] = ad.raw_blockdepth;
+              if (ad.raw_blockdepth > thread_max_raw_depth[current_thread])
+                thread_max_raw_depth[current_thread] = ad.raw_blockdepth;
+            }
+            
+            // Log the last thread
+            if (!thread_assignments[current_thread].empty())
+            {
+              const std::uint64_t min_depth = thread_min_raw_depth[current_thread] == std::numeric_limits<std::uint64_t>::max() ? 0 : thread_min_raw_depth[current_thread];
+              const std::uint64_t max_depth = thread_max_raw_depth[current_thread];
+              MINFO("Thread " << current_thread << " added " << thread_assignments[current_thread].size() 
+                    << " addresses with true block depths " << min_depth << " to " << max_depth);
+            }
+          }
+          
+          // Create threads with assigned accounts
+          for (std::size_t i = 0; i < queues.size(); ++i)
+          {
+            queues[i] = std::make_shared<rpc::scanner::queue>();
+
+            auto data = std::make_shared<thread_data>(
+              MONERO_UNWRAP(ctx.connect()), disk.clone(), std::move(thread_assignments[i]), queues[i], opts
+            );
+            threads.emplace_back(attrs, std::bind(&do_scan_loop, std::ref(self), std::move(data), i));
+          }
+        }
+        else
+        {
+          // Original algorithm: sort by height and divide evenly by count
+          std::sort(users.begin(), users.end(), by_height{});
+
+          for (std::size_t i = 0; i < queues.size(); ++i)
+          {
+            queues[i] = std::make_shared<rpc::scanner::queue>();
+
+            // this can create threads with no active accounts, they just wait
+            const std::size_t count = users.size() / (queues.size() - i);
+            std::vector<lws::account> thread_users{
+              std::make_move_iterator(users.end() - count), std::make_move_iterator(users.end())
+            };
+            users.erase(users.end() - count, users.end());
+
+            auto data = std::make_shared<thread_data>(
+              MONERO_UNWRAP(ctx.connect()), disk.clone(), std::move(thread_users), queues[i], opts
+            );
+            threads.emplace_back(attrs, std::bind(&do_scan_loop, std::ref(self), std::move(data), i));
+          }
         }
 
         users.clear();
@@ -1074,7 +1293,8 @@ namespace lws
           MONERO_UNWRAP(ctx.connect()),
           queues,
           std::move(active),
-          self.webhooks_.ssl_context()
+          self.webhooks_.ssl_context(),
+          opts.balance_new_addresses
         );
 
         rpc::scanner::server::start_user_checking(server);

src/scanner.h

@@ -43,11 +43,18 @@
 
 namespace lws
 {
+  constexpr std::uint64_t MINIMUM_BLOCK_DEPTH = 16;
+
   struct scanner_options
   {
+    double split_sync_threads;
+    std::uint64_t split_sync_depth;
+    std::uint64_t min_block_depth;
     std::uint32_t max_subaddresses;
     bool untrusted_daemon;
     bool regtest;
+    bool block_depth_threading;
+    bool balance_new_addresses;
   };
 
   //! Used in `scan_loop` by server

src/server_main.cpp

@@ -87,6 +87,11 @@ namespace
     const command_line::arg_descriptor<bool> regtest;
     const command_line::arg_descriptor<bool> version;
     const command_line::arg_descriptor<bool> auto_accept_import;
+    const command_line::arg_descriptor<bool> block_depth_threading;
+    const command_line::arg_descriptor<std::uint64_t> min_block_depth;
+    const command_line::arg_descriptor<double> split_sync_threads;
+    const command_line::arg_descriptor<std::uint64_t> split_sync_depth;
+    const command_line::arg_descriptor<bool> balance_new_addresses;
 
     static std::string get_default_zmq()
     {
@@ -137,6 +142,11 @@ namespace
       , regtest{"regtest", "Run in a regression testing mode", false}
       , version{"version", "Display version and quit", false}
       , auto_accept_import{"auto-accept-import", "Account import requests are automatically accepted", false}
+      , block_depth_threading{"block-depth-threading", "Balance thread workload by block depth instead of account count", false}
+      , min_block_depth{"min-block-depth", "Minimum block depth for block depth threading (defaults to 16)", lws::MINIMUM_BLOCK_DEPTH}
+      , split_sync_threads{"split-sync-threads", "Percentage of threads to use for fully synced accounts (0-1, requires --block-depth-threading, 0 to disable)", 0.0}
+      , split_sync_depth{"split-sync-depth", "Maximum block depth for an address to be considered synced (defaults to 10)", 10}
+      , balance_new_addresses{"balance-new-addresses", "Assign new addresses to thread with highest blockheight (or fewest addresses if tied)", false}
     {}
 
     void prepare(boost::program_options::options_description& description) const
@@ -175,6 +185,11 @@ namespace
       command_line::add_arg(description, regtest);
       command_line::add_arg(description, version);
       command_line::add_arg(description, auto_accept_import);
+      command_line::add_arg(description, block_depth_threading);
+      command_line::add_arg(description, min_block_depth);
+      command_line::add_arg(description, split_sync_threads);
+      command_line::add_arg(description, split_sync_depth);
+      command_line::add_arg(description, balance_new_addresses);
     }
   };
 
@@ -193,9 +208,14 @@ namespace
     std::string webhook_ssl_verification;
     std::chrono::minutes rates_interval;
     std::size_t scan_threads;
+    std::uint64_t split_sync_depth;
+    std::uint64_t min_block_depth;
     unsigned create_queue_max;
+    double split_sync_threads;
     bool untrusted_daemon;
     bool regtest;
+    bool block_depth_threading;
+    bool balance_new_addresses;
   };
 
   void print_version(std::ostream& out)
@@ -296,14 +316,25 @@ namespace
       command_line::get_arg(args, opts.webhook_ssl_verification),
       std::chrono::minutes{command_line::get_arg(args, opts.rates_interval)},
       command_line::get_arg(args, opts.scan_threads),
+      command_line::get_arg(args, opts.split_sync_depth),
+      command_line::get_arg(args, opts.min_block_depth),
       command_line::get_arg(args, opts.create_queue_max),
+      command_line::get_arg(args, opts.split_sync_threads),
       command_line::get_arg(args, opts.untrusted_daemon),
-      command_line::get_arg(args, opts.regtest)
+      command_line::get_arg(args, opts.regtest),
+      command_line::get_arg(args, opts.block_depth_threading),
+      command_line::get_arg(args, opts.balance_new_addresses)
     };
 
     if (prog.regtest && lws::config::network != cryptonote::MAINNET)
       MONERO_THROW(lws::error::configuration, "Regtest cannot be used with testnet or stagenet");
 
+    if (prog.split_sync_threads < 0.0 || prog.split_sync_threads > 1.0)
+      MONERO_THROW(lws::error::configuration, "--split-sync-threads must be between 0 and 1");
+    
+    if (prog.split_sync_threads > 0.0 && !prog.block_depth_threading)
+      MONERO_THROW(lws::error::configuration, "--split-sync-threads requires --block-depth-threading=true");
+
     if (!prog.lws_server_addr.empty() && (prog.rest_config.max_subaddresses || prog.untrusted_daemon))
       MONERO_THROW(lws::error::configuration, "Remote scanning cannot be used with subaddresses or untrusted daemon");
 
@@ -350,7 +381,7 @@ namespace
       prog.scan_threads,
       std::move(prog.lws_server_addr),
       std::move(prog.lws_server_pass),
-      lws::scanner_options{prog.rest_config.max_subaddresses, prog.untrusted_daemon, prog.regtest}
+      lws::scanner_options{prog.split_sync_threads, prog.split_sync_depth, prog.min_block_depth, prog.rest_config.max_subaddresses, prog.untrusted_daemon, prog.regtest, prog.block_depth_threading, prog.balance_new_addresses}
     );
   }
 } // anonymous

tests/unit/scanner.test.cpp

@@ -538,7 +538,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
       bmessage.output_indices.resize(1);
       messages.push_back(daemon_response(bmessage));
       {
-        static constexpr const lws::scanner_options opts{1, false};
+        static constexpr const lws::scanner_options opts{0, 0, lws::MINIMUM_BLOCK_DEPTH, 1, false, false, false, false};
         lws::scanner scanner{db.clone(), epee::net_utils::ssl_verification_t::none};
         boost::thread server_thread(&scanner_thread, std::ref(scanner), rpc.zmq_context(), std::cref(messages));
         const join on_scope_exit{server_thread};
@@ -864,7 +864,7 @@ LWS_CASE("lws::scanner::sync and lws::scanner::run")
       bmessage.output_indices.resize(1);
       messages.push_back(daemon_response(bmessage));
       {
-        static constexpr const lws::scanner_options opts{10, false};
+        static constexpr const lws::scanner_options opts{0, 0, lws::MINIMUM_BLOCK_DEPTH, 10, false, false, false, false};
         lws::scanner scanner{db.clone(), epee::net_utils::ssl_verification_t::none};
         boost::thread server_thread(&scanner_thread, std::ref(scanner), rpc.zmq_context(), std::cref(messages));
         const join on_scope_exit{server_thread};