Merge branch 'develop' into ximinez/online-delete-gaps

* upstream/develop:
  ci: Add dependabot config (6379)
  Fix tautological assertion (6393)
  chore: Apply clang-format width 100 (6387)

.clang-tidy

@@ -1,143 +1,105 @@
 ---
 Checks: "-*,
-  bugprone-argument-comment,
-  bugprone-assert-side-effect,
-  bugprone-bad-signal-to-kill-thread,
-  bugprone-bool-pointer-implicit-conversion,
-  bugprone-casting-through-void,
-  bugprone-chained-comparison,
-  bugprone-compare-pointer-to-member-virtual-function,
-  bugprone-copy-constructor-init,
-  bugprone-dangling-handle,
-  bugprone-dynamic-static-initializers,
-  bugprone-fold-init-type,
-  bugprone-forward-declaration-namespace,
-  bugprone-inaccurate-erase,
-  bugprone-incorrect-enable-if,
-  bugprone-incorrect-roundings,
-  bugprone-infinite-loop,
-  bugprone-integer-division,
-  bugprone-lambda-function-name,
-  bugprone-macro-parentheses,
-  bugprone-macro-repeated-side-effects,
-  bugprone-misplaced-operator-in-strlen-in-alloc,
-  bugprone-misplaced-pointer-arithmetic-in-alloc,
-  bugprone-misplaced-widening-cast,
-  bugprone-multi-level-implicit-pointer-conversion,
-  bugprone-multiple-new-in-one-expression,
-  bugprone-multiple-statement-macro,
-  bugprone-no-escape,
-  bugprone-non-zero-enum-to-bool-conversion,
-  bugprone-parent-virtual-call,
-  bugprone-posix-return,
-  bugprone-redundant-branch-condition,
-  bugprone-shared-ptr-array-mismatch,
-  bugprone-signal-handler,
-  bugprone-signed-char-misuse,
-  bugprone-sizeof-container,
-  bugprone-spuriously-wake-up-functions,
-  bugprone-standalone-empty,
-  bugprone-string-constructor,
-  bugprone-string-integer-assignment,
-  bugprone-string-literal-with-embedded-nul,
-  bugprone-stringview-nullptr,
-  bugprone-suspicious-enum-usage,
-  bugprone-suspicious-include,
-  bugprone-suspicious-memory-comparison,
-  bugprone-suspicious-memset-usage,
-  bugprone-suspicious-realloc-usage,
-  bugprone-suspicious-semicolon,
-  bugprone-suspicious-string-compare,
-  bugprone-swapped-arguments,
-  bugprone-terminating-continue,
-  bugprone-throw-keyword-missing,
-  bugprone-undefined-memory-manipulation,
-  bugprone-undelegated-constructor,
-  bugprone-unhandled-exception-at-new,
-  bugprone-unique-ptr-array-mismatch,
-  bugprone-unsafe-functions,
-  bugprone-virtual-near-miss,
-  cppcoreguidelines-no-suspend-with-lock,
-  cppcoreguidelines-virtual-class-destructor,
-  hicpp-ignored-remove-result,
-  misc-definitions-in-headers,
-  misc-header-include-cycle,
-  misc-misplaced-const,
-  misc-static-assert,
-  misc-throw-by-value-catch-by-reference,
-  misc-unused-alias-decls,
-  misc-unused-using-decls,
-  readability-duplicate-include,
-  readability-enum-initial-value,
-  readability-misleading-indentation,
-  readability-non-const-parameter,
-  readability-redundant-declaration,
-  readability-reference-to-constructed-temporary,
-  modernize-deprecated-headers,
-  modernize-make-shared,
-  modernize-make-unique,
-  performance-implicit-conversion-in-loop,
-  performance-move-constructor-init,
-  performance-trivially-destructible
+  bugprone-argument-comment
   "
-# ---
-# checks that have some issues that need to be resolved:
-#
-# bugprone-empty-catch,
+# bugprone-assert-side-effect,
+# bugprone-bad-signal-to-kill-thread,
+# bugprone-bool-pointer-implicit-conversion,
+# bugprone-casting-through-void,
+# bugprone-chained-comparison,
+# bugprone-compare-pointer-to-member-virtual-function,
+# bugprone-copy-constructor-init,
 # bugprone-crtp-constructor-accessibility,
+# bugprone-dangling-handle,
+# bugprone-dynamic-static-initializers,
+# bugprone-empty-catch,
+# bugprone-fold-init-type,
+# bugprone-forward-declaration-namespace,
+# bugprone-inaccurate-erase,
 # bugprone-inc-dec-in-conditions,
-# bugprone-reserved-identifier,
+# bugprone-incorrect-enable-if,
+# bugprone-incorrect-roundings,
+# bugprone-infinite-loop,
+# bugprone-integer-division,
+# bugprone-lambda-function-name,
+# bugprone-macro-parentheses,
+# bugprone-macro-repeated-side-effects,
+# bugprone-misplaced-operator-in-strlen-in-alloc,
+# bugprone-misplaced-pointer-arithmetic-in-alloc,
+# bugprone-misplaced-widening-cast,
 # bugprone-move-forwarding-reference,
-# bugprone-unused-local-non-trivial-variable,
-# bugprone-return-const-ref-from-parameter,
-# bugprone-switch-missing-default-case,
-# bugprone-sizeof-expression,
-# bugprone-suspicious-stringview-data-usage,
-# bugprone-suspicious-missing-comma,
-# bugprone-pointer-arithmetic-on-polymorphic-object,
+# bugprone-multi-level-implicit-pointer-conversion,
+# bugprone-multiple-new-in-one-expression,
+# bugprone-multiple-statement-macro,
+# bugprone-no-escape,
+# bugprone-non-zero-enum-to-bool-conversion,
 # bugprone-optional-value-conversion,
+# bugprone-parent-virtual-call,
+# bugprone-pointer-arithmetic-on-polymorphic-object,
+# bugprone-posix-return,
+# bugprone-redundant-branch-condition,
+# bugprone-reserved-identifier,
+# bugprone-return-const-ref-from-parameter,
+# bugprone-shared-ptr-array-mismatch,
+# bugprone-signal-handler,
+# bugprone-signed-char-misuse,
+# bugprone-sizeof-container,
+# bugprone-sizeof-expression,
+# bugprone-spuriously-wake-up-functions,
+# bugprone-standalone-empty,
+# bugprone-string-constructor,
+# bugprone-string-integer-assignment,
+# bugprone-string-literal-with-embedded-nul,
+# bugprone-stringview-nullptr,
+# bugprone-suspicious-enum-usage,
+# bugprone-suspicious-include,
+# bugprone-suspicious-memory-comparison,
+# bugprone-suspicious-memset-usage,
+# bugprone-suspicious-missing-comma,
+# bugprone-suspicious-realloc-usage,
+# bugprone-suspicious-semicolon,
+# bugprone-suspicious-string-compare,
+# bugprone-suspicious-stringview-data-usage,
+# bugprone-swapped-arguments,
+# bugprone-switch-missing-default-case,
+# bugprone-terminating-continue,
+# bugprone-throw-keyword-missing,
 # bugprone-too-small-loop-variable,
+# bugprone-undefined-memory-manipulation,
+# bugprone-undelegated-constructor,
+# bugprone-unhandled-exception-at-new,
+# bugprone-unhandled-self-assignment,
+# bugprone-unique-ptr-array-mismatch,
+# bugprone-unsafe-functions,
+# bugprone-unused-local-non-trivial-variable,
+# bugprone-unused-raii,
 # bugprone-unused-return-value,
 # bugprone-use-after-move,
-# bugprone-unhandled-self-assignment,
-# bugprone-unused-raii,
-#
-# cppcoreguidelines-misleading-capture-default-by-value,
+# bugprone-virtual-near-miss,
 # cppcoreguidelines-init-variables,
+# cppcoreguidelines-misleading-capture-default-by-value,
+# cppcoreguidelines-no-suspend-with-lock,
 # cppcoreguidelines-pro-type-member-init,
 # cppcoreguidelines-pro-type-static-cast-downcast,
-# cppcoreguidelines-use-default-member-init,
 # cppcoreguidelines-rvalue-reference-param-not-moved,
-#
+# cppcoreguidelines-use-default-member-init,
+# cppcoreguidelines-virtual-class-destructor,
+# hicpp-ignored-remove-result,
 # llvm-namespace-comment,
 # misc-const-correctness,
+# misc-definitions-in-headers,
+# misc-header-include-cycle,
 # misc-include-cleaner,
+# misc-misplaced-const,
 # misc-redundant-expression,
-#
-# readability-avoid-nested-conditional-operator,
-# readability-avoid-return-with-void-value,
-# readability-braces-around-statements,
-# readability-container-contains,
-# readability-container-size-empty,
-# readability-convert-member-functions-to-static,
-# readability-const-return-type,
-# readability-else-after-return,
-# readability-implicit-bool-conversion,
-# readability-inconsistent-declaration-parameter-name,
-# readability-identifier-naming,
-# readability-make-member-function-const,
-# readability-math-missing-parentheses,
-# readability-redundant-inline-specifier,
-# readability-redundant-member-init,
-# readability-redundant-casting,
-# readability-redundant-string-init,
-# readability-simplify-boolean-expr,
-# readability-static-definition-in-anonymous-namespace,
-# readability-suspicious-call-argument,
-# readability-use-std-min-max,
-# readability-static-accessed-through-instance,
-#
+# misc-static-assert,
+# misc-throw-by-value-catch-by-reference,
+# misc-unused-alias-decls,
+# misc-unused-using-decls,
 # modernize-concat-nested-namespaces,
+# modernize-deprecated-headers,
+# modernize-make-shared,
+# modernize-make-unique,
 # modernize-pass-by-value,
 # modernize-type-traits,
 # modernize-use-designated-initializers,
@@ -149,50 +111,79 @@ Checks: "-*,
 # modernize-use-starts-ends-with,
 # modernize-use-std-numbers,
 # modernize-use-using,
-#
 # performance-faster-string-find,
 # performance-for-range-copy,
+# performance-implicit-conversion-in-loop,
 # performance-inefficient-vector-operation,
 # performance-move-const-arg,
+# performance-move-constructor-init,
 # performance-no-automatic-move,
-# ---
+# performance-trivially-destructible,
+# readability-avoid-nested-conditional-operator,
+# readability-avoid-return-with-void-value,
+# readability-braces-around-statements,
+# readability-const-return-type,
+# readability-container-contains,
+# readability-container-size-empty,
+# readability-convert-member-functions-to-static,
+# readability-duplicate-include,
+# readability-else-after-return,
+# readability-enum-initial-value,
+# readability-implicit-bool-conversion,
+# readability-inconsistent-declaration-parameter-name,
+# readability-identifier-naming,
+# readability-make-member-function-const,
+# readability-math-missing-parentheses,
+# readability-misleading-indentation,
+# readability-non-const-parameter,
+# readability-redundant-casting,
+# readability-redundant-declaration,
+# readability-redundant-inline-specifier,
+# readability-redundant-member-init,
+# readability-redundant-string-init,
+# readability-reference-to-constructed-temporary,
+# readability-simplify-boolean-expr,
+# readability-static-accessed-through-instance,
+# readability-static-definition-in-anonymous-namespace,
+# readability-suspicious-call-argument,
+# readability-use-std-min-max
 #
-CheckOptions:
-  #   readability-braces-around-statements.ShortStatementLines: 2
-  #   readability-identifier-naming.MacroDefinitionCase: UPPER_CASE
-  #   readability-identifier-naming.ClassCase: CamelCase
-  #   readability-identifier-naming.StructCase: CamelCase
-  #   readability-identifier-naming.UnionCase: CamelCase
-  #   readability-identifier-naming.EnumCase: CamelCase
-  #   readability-identifier-naming.EnumConstantCase: CamelCase
-  #   readability-identifier-naming.ScopedEnumConstantCase: CamelCase
-  #   readability-identifier-naming.GlobalConstantCase: UPPER_CASE
-  #   readability-identifier-naming.GlobalConstantPrefix: "k"
-  #   readability-identifier-naming.GlobalVariableCase: CamelCase
-  #   readability-identifier-naming.GlobalVariablePrefix: "g"
-  #   readability-identifier-naming.ConstexprFunctionCase: camelBack
-  #   readability-identifier-naming.ConstexprMethodCase: camelBack
-  #   readability-identifier-naming.ClassMethodCase: camelBack
-  #   readability-identifier-naming.ClassMemberCase: camelBack
-  #   readability-identifier-naming.ClassConstantCase: UPPER_CASE
-  #   readability-identifier-naming.ClassConstantPrefix: "k"
-  #   readability-identifier-naming.StaticConstantCase: UPPER_CASE
-  #   readability-identifier-naming.StaticConstantPrefix: "k"
-  #   readability-identifier-naming.StaticVariableCase: UPPER_CASE
-  #   readability-identifier-naming.StaticVariablePrefix: "k"
-  #   readability-identifier-naming.ConstexprVariableCase: UPPER_CASE
-  #   readability-identifier-naming.ConstexprVariablePrefix: "k"
-  #   readability-identifier-naming.LocalConstantCase: camelBack
-  #   readability-identifier-naming.LocalVariableCase: camelBack
-  #   readability-identifier-naming.TemplateParameterCase: CamelCase
-  #   readability-identifier-naming.ParameterCase: camelBack
-  #   readability-identifier-naming.FunctionCase: camelBack
-  #   readability-identifier-naming.MemberCase: camelBack
-  #   readability-identifier-naming.PrivateMemberSuffix: _
-  #   readability-identifier-naming.ProtectedMemberSuffix: _
-  #   readability-identifier-naming.PublicMemberSuffix: ""
-  #   readability-identifier-naming.FunctionIgnoredRegexp: ".*tag_invoke.*"
-  bugprone-unsafe-functions.ReportMoreUnsafeFunctions: true
+# CheckOptions:
+#   readability-braces-around-statements.ShortStatementLines: 2
+#   readability-identifier-naming.MacroDefinitionCase: UPPER_CASE
+#   readability-identifier-naming.ClassCase: CamelCase
+#   readability-identifier-naming.StructCase: CamelCase
+#   readability-identifier-naming.UnionCase: CamelCase
+#   readability-identifier-naming.EnumCase: CamelCase
+#   readability-identifier-naming.EnumConstantCase: CamelCase
+#   readability-identifier-naming.ScopedEnumConstantCase: CamelCase
+#   readability-identifier-naming.GlobalConstantCase: UPPER_CASE
+#   readability-identifier-naming.GlobalConstantPrefix: "k"
+#   readability-identifier-naming.GlobalVariableCase: CamelCase
+#   readability-identifier-naming.GlobalVariablePrefix: "g"
+#   readability-identifier-naming.ConstexprFunctionCase: camelBack
+#   readability-identifier-naming.ConstexprMethodCase: camelBack
+#   readability-identifier-naming.ClassMethodCase: camelBack
+#   readability-identifier-naming.ClassMemberCase: camelBack
+#   readability-identifier-naming.ClassConstantCase: UPPER_CASE
+#   readability-identifier-naming.ClassConstantPrefix: "k"
+#   readability-identifier-naming.StaticConstantCase: UPPER_CASE
+#   readability-identifier-naming.StaticConstantPrefix: "k"
+#   readability-identifier-naming.StaticVariableCase: UPPER_CASE
+#   readability-identifier-naming.StaticVariablePrefix: "k"
+#   readability-identifier-naming.ConstexprVariableCase: UPPER_CASE
+#   readability-identifier-naming.ConstexprVariablePrefix: "k"
+#   readability-identifier-naming.LocalConstantCase: camelBack
+#   readability-identifier-naming.LocalVariableCase: camelBack
+#   readability-identifier-naming.TemplateParameterCase: CamelCase
+#   readability-identifier-naming.ParameterCase: camelBack
+#   readability-identifier-naming.FunctionCase: camelBack
+#   readability-identifier-naming.MemberCase: camelBack
+#   readability-identifier-naming.PrivateMemberSuffix: _
+#   readability-identifier-naming.ProtectedMemberSuffix: _
+#   readability-identifier-naming.PublicMemberSuffix: ""
+#   readability-identifier-naming.FunctionIgnoredRegexp: ".*tag_invoke.*"
+#   bugprone-unsafe-functions.ReportMoreUnsafeFunctions: true
 #   bugprone-unused-return-value.CheckedReturnTypes: ::std::error_code;::std::error_condition;::std::errc
 #   misc-include-cleaner.IgnoreHeaders: '.*/(detail|impl)/.*;.*(expected|unexpected).*;.*ranges_lower_bound\.h;time.h;stdlib.h;__chrono/.*;fmt/chrono.h;boost/uuid/uuid_hash.hpp'
 #

.github/workflows/reusable-build-test-config.yml

@@ -101,7 +101,7 @@ jobs:
     steps:
       - name: Cleanup workspace (macOS and Windows)
         if: ${{ runner.os == 'macOS' || runner.os == 'Windows' }}
-        uses: XRPLF/actions/cleanup-workspace@c7d9ce5ebb03c752a354889ecd870cadfc2b1cd4
+        uses: XRPLF/actions/cleanup-workspace@cf0433aa74563aead044a1e395610c96d65a37cf
 
       - name: Checkout repository
         uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
@@ -229,21 +229,8 @@ jobs:
         env:
           BUILD_NPROC: ${{ steps.nproc.outputs.nproc }}
         run: |
-          set -o pipefail
-          ./xrpld --unittest --unittest-jobs "${BUILD_NPROC}" 2>&1 | tee unittest.log
+          ./xrpld --unittest --unittest-jobs "${BUILD_NPROC}"
 
-      - name: Show test failure summary
-        if: ${{ failure() && !inputs.build_only }}
-        working-directory: ${{ runner.os == 'Windows' && format('{0}/{1}', env.BUILD_DIR, inputs.build_type) || env.BUILD_DIR }}
-        run: |
-          if [ ! -f unittest.log ]; then
-            echo "unittest.log not found; embedded tests may not have run."
-            exit 0
-          fi
-
-          if ! grep -E "failed" unittest.log; then
-            echo "Log present but no failure lines found in unittest.log."
-          fi
       - name: Debug failure (Linux)
         if: ${{ failure() && runner.os == 'Linux' && !inputs.build_only }}
         run: |

.github/workflows/reusable-clang-tidy-files.yml

@@ -78,9 +78,9 @@ jobs:
         id: run_clang_tidy
         continue-on-error: true
         env:
-          TARGETS: ${{ inputs.files != '' && inputs.files || 'src tests' }}
+          FILES: ${{ inputs.files }}
         run: |
-          run-clang-tidy -j ${{ steps.nproc.outputs.nproc }} -p "${BUILD_DIR}" ${TARGETS} 2>&1 | tee clang-tidy-output.txt
+          run-clang-tidy -j ${{ steps.nproc.outputs.nproc }} -p "$BUILD_DIR" $FILES 2>&1 | tee clang-tidy-output.txt
 
       - name: Upload clang-tidy output
         if: steps.run_clang_tidy.outcome != 'success'

.github/workflows/reusable-clang-tidy.yml

@@ -22,8 +22,7 @@ jobs:
     if: ${{ inputs.check_only_changed }}
     runs-on: ubuntu-latest
     outputs:
-      clang_tidy_config_changed: ${{ steps.changed_clang_tidy.outputs.any_changed }}
-      any_cpp_changed: ${{ steps.changed_files.outputs.any_changed }}
+      any_changed: ${{ steps.changed_files.outputs.any_changed }}
       all_changed_files: ${{ steps.changed_files.outputs.all_changed_files }}
     steps:
       - name: Checkout repository
@@ -39,17 +38,10 @@ jobs:
             **/*.ipp
           separator: " "
 
-      - name: Get changed clang-tidy configuration
-        id: changed_clang_tidy
-        uses: tj-actions/changed-files@7dee1b0c1557f278e5c7dc244927139d78c0e22a # v47.0.4
-        with:
-          files: |
-            .clang-tidy
-
   run-clang-tidy:
     needs: [determine-files]
-    if: ${{ always() && !cancelled() && (!inputs.check_only_changed || needs.determine-files.outputs.any_cpp_changed == 'true' || needs.determine-files.outputs.clang_tidy_config_changed == 'true') }}
+    if: ${{ always() && !cancelled() && (!inputs.check_only_changed || needs.determine-files.outputs.any_changed == 'true') }}
     uses: ./.github/workflows/reusable-clang-tidy-files.yml
     with:
-      files: ${{ (needs.determine-files.outputs.clang_tidy_config_changed == 'true' && '') || (inputs.check_only_changed && needs.determine-files.outputs.all_changed_files || '') }}
+      files: ${{ inputs.check_only_changed && needs.determine-files.outputs.all_changed_files || '' }}
       create_issue_on_failure: ${{ inputs.create_issue_on_failure }}

.github/workflows/upload-conan-deps.yml

@@ -64,7 +64,7 @@ jobs:
     steps:
       - name: Cleanup workspace (macOS and Windows)
         if: ${{ runner.os == 'macOS' || runner.os == 'Windows' }}
-        uses: XRPLF/actions/cleanup-workspace@c7d9ce5ebb03c752a354889ecd870cadfc2b1cd4
+        uses: XRPLF/actions/cleanup-workspace@cf0433aa74563aead044a1e395610c96d65a37cf
 
       - name: Checkout repository
         uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2

CONTRIBUTING.md

@@ -251,29 +251,6 @@ pip3 install pre-commit
 pre-commit install
 ```
 
-## Clang-tidy
-
-All code must pass `clang-tidy` checks according to the settings in [`.clang-tidy`](./.clang-tidy).
-
-There is a Continuous Integration job that runs clang-tidy on pull requests. The CI will check:
-
-- All changed C++ files (`.cpp`, `.h`, `.ipp`) when only code files are modified
-- **All files in the repository** when the `.clang-tidy` configuration file is changed
-
-This ensures that configuration changes don't introduce new warnings across the codebase.
-
-### Running clang-tidy locally
-
-Before running clang-tidy, you must build the project to generate required files (particularly protobuf headers). Refer to [`BUILD.md`](./BUILD.md) for build instructions.
-
-Then run clang-tidy on your local changes:
-
-```
-run-clang-tidy -p build src tests
-```
-
-This will check all source files in the `src` and `tests` directories using the compile commands from your `build` directory.
-
 ## Contracts and instrumentation
 
 We are using [Antithesis](https://antithesis.com/) for continuous fuzzing,

cfg/xrpld-example.cfg

@@ -1033,10 +1033,11 @@
 #                           The online delete process checks periodically
 #                           that xrpld is still in sync with the network,
 #                           and that the validated ledger is less than
-#                           'age_threshold_seconds' old. If not, then continue
+#                           'age_threshold_seconds' old, and that all
+#                           recent ledgers are available. If not, then continue
 #                           sleeping for this number of seconds and
 #                           checking until healthy.
-#                           Default is 5.
+#                           Default is 1.
 #
 #   Notes:
 #       The 'node_db' entry configures the primary, persistent storage.

cspell.config.yaml

@@ -71,14 +71,12 @@ words:
   - coldwallet
   - compr
   - conanfile
-  - cppcoro
   - conanrun
   - confs
   - connectability
   - coro
   - coros
   - cowid
-  - cppcoro
   - cryptocondition
   - cryptoconditional
   - cryptoconditions
@@ -101,14 +99,11 @@ words:
   - endmacro
   - exceptioned
   - Falco
-  - fcontext
   - finalizers
   - firewalled
-  - fcontext
   - fmtdur
   - fsanitize
   - funclets
-  - gantt
   - gcov
   - gcovr
   - ghead
@@ -190,7 +185,6 @@ words:
   - ostr
   - pargs
   - partitioner
-  - pratik
   - paychan
   - paychans
   - permdex
@@ -198,7 +192,6 @@ words:
   - permissioned
   - pointee
   - populator
-  - pratik
   - preauth
   - preauthorization
   - preauthorize
@@ -213,7 +206,6 @@ words:
   - queuable
   - Raphson
   - replayer
-  - repost
   - rerere
   - retriable
   - RIPD
@@ -244,7 +236,6 @@ words:
   - soci
   - socidb
   - sslws
-  - stackful
   - statsd
   - STATSDCOLLECTOR
   - stissue

include/xrpl/core/CoroTask.h

@@ -1,687 +0,0 @@
-#pragma once
-
-#include <coroutine>
-#include <exception>
-#include <utility>
-#include <variant>
-
-namespace xrpl {
-
-template <typename T = void>
-class CoroTask;
-
-/**
- * CoroTask<void> -- coroutine return type for void-returning coroutines.
- *
- * Class / Dependency Diagram
- * ==========================
- *
- *   CoroTask<void>
- *   +-----------------------------------------------+
- *   | - handle_ : Handle (coroutine_handle<promise>) |
- *   +-----------------------------------------------+
- *   | + handle(), done()                             |
- *   | + await_ready/suspend/resume  (Awaiter iface)  |
- *   +-----------------------------------------------+
- *            |  owns
- *            v
- *   promise_type
- *   +-----------------------------------------------+
- *   | - exception_    : std::exception_ptr           |
- *   | - continuation_ : std::coroutine_handle<>      |
- *   +-----------------------------------------------+
- *   | + get_return_object() -> CoroTask              |
- *   | + initial_suspend()   -> suspend_always (lazy) |
- *   | + final_suspend()     -> FinalAwaiter          |
- *   | + return_void()                                |
- *   | + unhandled_exception()                        |
- *   +-----------------------------------------------+
- *            |  returns at final_suspend
- *            v
- *   FinalAwaiter
- *   +-----------------------------------------------+
- *   | await_suspend(h):                              |
- *   |   if continuation_ set -> symmetric transfer   |
- *   |   else                 -> noop_coroutine        |
- *   +-----------------------------------------------+
- *
- * Design Notes
- * ------------
- * - Lazy start: initial_suspend returns suspend_always, so the coroutine
- *   body does not execute until the handle is explicitly resumed.
- * - Symmetric transfer: await_suspend returns a coroutine_handle instead
- *   of void/bool, allowing the scheduler to jump directly to the next
- *   coroutine without growing the call stack.
- * - Continuation chaining: when one CoroTask is co_await-ed inside
- *   another, the caller's handle is stored as continuation_ so
- *   FinalAwaiter can resume it when this task finishes.
- * - Move-only: the handle is exclusively owned; copy is deleted.
- *
- * Usage Examples
- * ==============
- *
- * 1. Basic void coroutine (the most common case in rippled):
- *
- *      CoroTask<void> doWork(std::shared_ptr<CoroTaskRunner> runner) {
- *          // do something
- *          co_await runner->suspend();   // yield control
- *          // resumed later via runner->post() or runner->resume()
- *          co_return;
- *      }
- *
- * 2. co_await-ing one CoroTask<void> from another (chaining):
- *
- *      CoroTask<void> inner() {
- *          // ...
- *          co_return;
- *      }
- *      CoroTask<void> outer() {
- *          co_await inner();   // continuation_ links outer -> inner
- *          co_return;          // FinalAwaiter resumes outer
- *      }
- *
- * 3. Exceptions propagate through co_await:
- *
- *      CoroTask<void> failing() {
- *          throw std::runtime_error("oops");
- *          co_return;
- *      }
- *      CoroTask<void> caller() {
- *          try { co_await failing(); }
- *          catch (std::runtime_error const&) { // caught here }
- *      }
- *
- * Caveats / Pitfalls
- * ==================
- *
- * BUG-RISK: Dangling references in coroutine parameters.
- *   Coroutine parameters are copied into the frame, but references
- *   are NOT -- they are stored as-is. If the referent goes out of scope
- *   before the coroutine finishes, you get use-after-free.
- *
- *      // BROKEN -- local dies before coroutine runs:
- *      CoroTask<void> bad(int& ref) { co_return; }
- *      void launch() {
- *          int local = 42;
- *          auto task = bad(local);  // frame stores &local
- *      }  // local destroyed; frame holds dangling ref
- *
- *      // FIX -- pass by value, or ensure lifetime via shared_ptr.
- *
- * BUG-RISK: GCC 14 corrupts reference captures in coroutine lambdas.
- *   When a lambda that returns CoroTask captures by reference ([&]),
- *   GCC 14 may generate a corrupted coroutine frame. Always capture
- *   by explicit pointer-to-value instead:
- *
- *      // BROKEN on GCC 14:
- *      jq.postCoroTask(t, n, [&](auto) -> CoroTask<void> { ... });
- *
- *      // FIX -- capture pointers explicitly:
- *      jq.postCoroTask(t, n, [ptr = &val](auto) -> CoroTask<void> { ... });
- *
- * BUG-RISK: Resuming a destroyed or completed CoroTask.
- *   Calling handle().resume() after the coroutine has already run to
- *   completion (done() == true) is undefined behavior. The CoroTaskRunner
- *   guards against this with an XRPL_ASSERT, but standalone usage of
- *   CoroTask must check done() before resuming.
- *
- * BUG-RISK: Moving a CoroTask that is being awaited.
- *   If task A is co_await-ed by task B (so A.continuation_ == B), moving
- *   or destroying A will invalidate the continuation link. Never move
- *   or reassign a CoroTask while it is mid-execution or being awaited.
- *
- * LIMITATION: CoroTask is fire-and-forget for the top-level owner.
- *   There is no built-in notification when the coroutine finishes.
- *   The caller must use external synchronization (e.g. CoroTaskRunner::join
- *   or a gate/condition_variable) to know when it is done.
- *
- * LIMITATION: No cancellation token.
- *   There is no way to cancel a suspended CoroTask from outside. The
- *   coroutine body must cooperatively check a flag (e.g. jq_.isStopping())
- *   after each co_await and co_return early if needed.
- *
- * LIMITATION: Stackless -- cannot suspend from nested non-coroutine calls.
- *   If a coroutine calls a regular function that wants to "yield", it
- *   cannot. Only the immediate coroutine body can use co_await.
- *   This is acceptable for rippled because all yield() sites are shallow.
- */
-template <>
-class CoroTask<void>
-{
-public:
-    struct promise_type;
-    using Handle = std::coroutine_handle<promise_type>;
-
-    /**
-     * Coroutine promise. Compiler uses this to manage coroutine state.
-     * Stores the exception (if any) and the continuation handle for
-     * symmetric transfer back to the awaiting coroutine.
-     */
-    struct promise_type
-    {
-        // Captured exception from the coroutine body, rethrown in
-        // await_resume() when this task is co_await-ed by a caller.
-        std::exception_ptr exception_;
-
-        // Handle to the coroutine that is co_await-ing this task.
-        // Set by await_suspend(). FinalAwaiter uses it for symmetric
-        // transfer back to the caller. Null if this is a top-level task.
-        std::coroutine_handle<> continuation_;
-
-        /**
-         * Create the CoroTask return object.
-         * Called by the compiler at coroutine creation.
-         */
-        CoroTask
-        get_return_object()
-        {
-            return CoroTask{Handle::from_promise(*this)};
-        }
-
-        /**
-         * Lazy start. The coroutine body does not execute until the
-         * handle is explicitly resumed (e.g. by CoroTaskRunner::resume).
-         */
-        std::suspend_always
-        initial_suspend() noexcept
-        {
-            return {};
-        }
-
-        /**
-         * Awaiter returned by final_suspend(). Uses symmetric transfer:
-         * if a continuation exists, transfers control directly to it
-         * (tail-call, no stack growth). Otherwise returns noop_coroutine
-         * so the coroutine frame stays alive for the owner to destroy.
-         */
-        struct FinalAwaiter
-        {
-            /**
-             * Always false. We need await_suspend to run for
-             * symmetric transfer.
-             */
-            bool
-            await_ready() noexcept
-            {
-                return false;
-            }
-
-            /**
-             * Symmetric transfer: returns the continuation handle so
-             * the compiler emits a tail-call instead of a nested resume.
-             * If no continuation is set, returns noop_coroutine to
-             * suspend at final_suspend without destroying the frame.
-             *
-             * @param h Handle to this completing coroutine
-             *
-             * @return Continuation handle, or noop_coroutine
-             */
-            std::coroutine_handle<>
-            await_suspend(Handle h) noexcept
-            {
-                if (auto cont = h.promise().continuation_)
-                    return cont;
-                return std::noop_coroutine();
-            }
-
-            void
-            await_resume() noexcept
-            {
-            }
-        };
-
-        /**
-         * Returns FinalAwaiter for symmetric transfer at coroutine end.
-         */
-        FinalAwaiter
-        final_suspend() noexcept
-        {
-            return {};
-        }
-
-        /**
-         * Called by the compiler for `co_return;` (void coroutine).
-         */
-        void
-        return_void()
-        {
-        }
-
-        /**
-         * Called by the compiler when an exception escapes the coroutine
-         * body. Captures it for later rethrowing in await_resume().
-         */
-        void
-        unhandled_exception()
-        {
-            exception_ = std::current_exception();
-        }
-    };
-
-    /**
-     * Default constructor. Creates an empty (null handle) task.
-     */
-    CoroTask() = default;
-
-    /**
-     * Takes ownership of a compiler-generated coroutine handle.
-     *
-     * @param h Coroutine handle to own
-     */
-    explicit CoroTask(Handle h) : handle_(h)
-    {
-    }
-
-    /**
-     * Destroys the coroutine frame if this task owns one.
-     */
-    ~CoroTask()
-    {
-        if (handle_)
-            handle_.destroy();
-    }
-
-    /**
-     * Move constructor. Transfers handle ownership, leaves other empty.
-     */
-    CoroTask(CoroTask&& other) noexcept : handle_(std::exchange(other.handle_, {}))
-    {
-    }
-
-    /**
-     * Move assignment. Destroys current frame (if any), takes other's.
-     */
-    CoroTask&
-    operator=(CoroTask&& other) noexcept
-    {
-        if (this != &other)
-        {
-            if (handle_)
-                handle_.destroy();
-            handle_ = std::exchange(other.handle_, {});
-        }
-        return *this;
-    }
-
-    CoroTask(CoroTask const&) = delete;
-    CoroTask&
-    operator=(CoroTask const&) = delete;
-
-    /**
-     * @return The underlying coroutine_handle
-     */
-    Handle
-    handle() const
-    {
-        return handle_;
-    }
-
-    /**
-     * @return true if the coroutine has run to completion (or thrown)
-     */
-    bool
-    done() const
-    {
-        return handle_ && handle_.done();
-    }
-
-    // -- Awaiter interface: allows `co_await someCoroTask;` --
-
-    /**
-     * Always false. This task is lazy, so co_await always suspends
-     * the caller to set up the continuation link.
-     */
-    bool
-    await_ready() const noexcept
-    {
-        return false;
-    }
-
-    /**
-     * Stores the caller's handle as our continuation, then returns
-     * our handle for symmetric transfer (caller suspends, we resume).
-     *
-     * @param caller Handle of the coroutine doing co_await on us
-     *
-     * @return Our handle for symmetric transfer
-     */
-    std::coroutine_handle<>
-    await_suspend(std::coroutine_handle<> caller) noexcept
-    {
-        handle_.promise().continuation_ = caller;
-        return handle_;  // Symmetric transfer
-    }
-
-    /**
-     * Called when the caller resumes after co_await. Rethrows any
-     * exception captured by unhandled_exception().
-     */
-    void
-    await_resume()
-    {
-        if (auto& ep = handle_.promise().exception_)
-            std::rethrow_exception(ep);
-    }
-
-private:
-    // Exclusively-owned coroutine handle. Null after move or default
-    // construction. Destroyed in the destructor.
-    Handle handle_;
-};
-
-/**
- * CoroTask<T> -- coroutine return type for value-returning coroutines.
- *
- * Class / Dependency Diagram
- * ==========================
- *
- *   CoroTask<T>
- *   +-----------------------------------------------+
- *   | - handle_ : Handle (coroutine_handle<promise>) |
- *   +-----------------------------------------------+
- *   | + handle(), done()                             |
- *   | + await_ready/suspend/resume  (Awaiter iface)  |
- *   +-----------------------------------------------+
- *            |  owns
- *            v
- *   promise_type
- *   +-----------------------------------------------+
- *   | - result_       : variant<monostate, T,        |
- *   |                           exception_ptr>       |
- *   | - continuation_ : std::coroutine_handle<>      |
- *   +-----------------------------------------------+
- *   | + get_return_object() -> CoroTask              |
- *   | + initial_suspend()   -> suspend_always (lazy) |
- *   | + final_suspend()     -> FinalAwaiter          |
- *   | + return_value(T)     -> stores in result_[1]  |
- *   | + unhandled_exception -> stores in result_[2]  |
- *   +-----------------------------------------------+
- *            |  returns at final_suspend
- *            v
- *   FinalAwaiter  (same symmetric-transfer pattern as CoroTask<void>)
- *
- * Value Extraction
- * ----------------
- * await_resume() inspects the variant:
- *   - index 2 (exception_ptr) -> rethrow
- *   - index 1 (T)             -> return value via move
- *
- * Usage Examples
- * ==============
- *
- * 1. Simple value return:
- *
- *      CoroTask<int> computeAnswer() { co_return 42; }
- *
- *      CoroTask<void> caller() {
- *          int v = co_await computeAnswer();  // v == 42
- *      }
- *
- * 2. Chaining value-returning coroutines:
- *
- *      CoroTask<int> add(int a, int b) { co_return a + b; }
- *      CoroTask<int> doubleSum(int a, int b) {
- *          int s = co_await add(a, b);
- *          co_return s * 2;
- *      }
- *
- * 3. Exception propagation from inner to outer:
- *
- *      CoroTask<int> failing() {
- *          throw std::runtime_error("bad");
- *          co_return 0;  // never reached
- *      }
- *      CoroTask<void> caller() {
- *          try {
- *              int v = co_await failing();  // throws here
- *          } catch (std::runtime_error const& e) {
- *              // e.what() == "bad"
- *          }
- *      }
- *
- * Caveats / Pitfalls (in addition to CoroTask<void> caveats above)
- * ================================================================
- *
- * BUG-RISK: await_resume() moves the value out of the variant.
- *   Calling co_await on the same CoroTask<T> instance twice is undefined
- *   behavior -- the second call will see a moved-from T. CoroTask is
- *   single-shot: one co_return, one co_await.
- *
- * BUG-RISK: T must be move-constructible.
- *   return_value(T) takes by value and moves into the variant.
- *   Types that are not movable cannot be used as T.
- *
- * LIMITATION: No co_yield support.
- *   CoroTask<T> only supports a single co_return. It does not implement
- *   yield_value(), so using co_yield inside a CoroTask<T> coroutine is a
- *   compile error. For streaming values, a different return type
- *   (e.g. Generator<T>) would be needed.
- *
- * LIMITATION: Result is only accessible via co_await.
- *   There is no .get() or .result() method. The value can only be
- *   extracted by co_await-ing the CoroTask<T> from inside another
- *   coroutine. For extracting results in non-coroutine code, pass a
- *   pointer to the caller and write through it (as the tests do).
- */
-template <typename T>
-class CoroTask
-{
-public:
-    struct promise_type;
-    using Handle = std::coroutine_handle<promise_type>;
-
-    /**
-     * Coroutine promise for value-returning coroutines.
-     * Stores the result as a variant: monostate (not yet set),
-     * T (co_return value), or exception_ptr (unhandled exception).
-     */
-    struct promise_type
-    {
-        // Tri-state result:
-        //   index 0 (monostate) -- coroutine has not yet completed
-        //   index 1 (T)         -- co_return value stored here
-        //   index 2 (exception) -- unhandled exception captured here
-        std::variant<std::monostate, T, std::exception_ptr> result_;
-
-        // Handle to the coroutine co_await-ing this task. Used by
-        // FinalAwaiter for symmetric transfer. Null for top-level tasks.
-        std::coroutine_handle<> continuation_;
-
-        /**
-         * Create the CoroTask return object.
-         * Called by the compiler at coroutine creation.
-         */
-        CoroTask
-        get_return_object()
-        {
-            return CoroTask{Handle::from_promise(*this)};
-        }
-
-        /**
-         * Lazy start. Coroutine body does not run until explicitly resumed.
-         */
-        std::suspend_always
-        initial_suspend() noexcept
-        {
-            return {};
-        }
-
-        /**
-         * Symmetric-transfer awaiter at coroutine completion.
-         * Same pattern as CoroTask<void>::FinalAwaiter.
-         */
-        struct FinalAwaiter
-        {
-            bool
-            await_ready() noexcept
-            {
-                return false;
-            }
-
-            /**
-             * Returns continuation for symmetric transfer, or
-             * noop_coroutine if this is a top-level task.
-             *
-             * @param h Handle to this completing coroutine
-             *
-             * @return Continuation handle, or noop_coroutine
-             */
-            std::coroutine_handle<>
-            await_suspend(Handle h) noexcept
-            {
-                if (auto cont = h.promise().continuation_)
-                    return cont;
-                return std::noop_coroutine();
-            }
-
-            void
-            await_resume() noexcept
-            {
-            }
-        };
-
-        FinalAwaiter
-        final_suspend() noexcept
-        {
-            return {};
-        }
-
-        /**
-         * Called by the compiler for `co_return value;`.
-         * Moves the value into result_ at index 1.
-         *
-         * @param value The value to store
-         */
-        void
-        return_value(T value)
-        {
-            result_.template emplace<1>(std::move(value));
-        }
-
-        /**
-         * Captures unhandled exceptions at index 2 of result_.
-         * Rethrown later in await_resume().
-         */
-        void
-        unhandled_exception()
-        {
-            result_.template emplace<2>(std::current_exception());
-        }
-    };
-
-    /**
-     * Default constructor. Creates an empty (null handle) task.
-     */
-    CoroTask() = default;
-
-    /**
-     * Takes ownership of a compiler-generated coroutine handle.
-     *
-     * @param h Coroutine handle to own
-     */
-    explicit CoroTask(Handle h) : handle_(h)
-    {
-    }
-
-    /**
-     * Destroys the coroutine frame if this task owns one.
-     */
-    ~CoroTask()
-    {
-        if (handle_)
-            handle_.destroy();
-    }
-
-    /**
-     * Move constructor. Transfers handle ownership, leaves other empty.
-     */
-    CoroTask(CoroTask&& other) noexcept : handle_(std::exchange(other.handle_, {}))
-    {
-    }
-
-    /**
-     * Move assignment. Destroys current frame (if any), takes other's.
-     */
-    CoroTask&
-    operator=(CoroTask&& other) noexcept
-    {
-        if (this != &other)
-        {
-            if (handle_)
-                handle_.destroy();
-            handle_ = std::exchange(other.handle_, {});
-        }
-        return *this;
-    }
-
-    CoroTask(CoroTask const&) = delete;
-    CoroTask&
-    operator=(CoroTask const&) = delete;
-
-    /**
-     * @return The underlying coroutine_handle
-     */
-    Handle
-    handle() const
-    {
-        return handle_;
-    }
-
-    /**
-     * @return true if the coroutine has run to completion (or thrown)
-     */
-    bool
-    done() const
-    {
-        return handle_ && handle_.done();
-    }
-
-    // -- Awaiter interface: allows `T val = co_await someCoroTask;` --
-
-    /**
-     * Always false. co_await always suspends to set up continuation.
-     */
-    bool
-    await_ready() const noexcept
-    {
-        return false;
-    }
-
-    /**
-     * Stores caller as continuation, returns our handle for
-     * symmetric transfer.
-     *
-     * @param caller Handle of the coroutine doing co_await on us
-     *
-     * @return Our handle for symmetric transfer
-     */
-    std::coroutine_handle<>
-    await_suspend(std::coroutine_handle<> caller) noexcept
-    {
-        handle_.promise().continuation_ = caller;
-        return handle_;
-    }
-
-    /**
-     * Extracts the result: rethrows if exception, otherwise moves
-     * the T value out of the variant. Single-shot: calling twice
-     * on the same task is undefined (moved-from T).
-     *
-     * @return The co_return-ed value
-     */
-    T
-    await_resume()
-    {
-        auto& result = handle_.promise().result_;
-        if (auto* ep = std::get_if<2>(&result))
-            std::rethrow_exception(*ep);
-        return std::get<1>(std::move(result));
-    }
-
-private:
-    // Exclusively-owned coroutine handle. Null after move or default
-    // construction. Destroyed in the destructor.
-    Handle handle_;
-};
-
-}  // namespace xrpl

include/xrpl/core/CoroTaskRunner.ipp

@@ -1,329 +0,0 @@
-#pragma once
-
-/**
- * @file CoroTaskRunner.ipp
- *
- * CoroTaskRunner inline implementation.
- *
- * This file contains the business logic for managing C++20 coroutines
- * on the JobQueue. It is included at the bottom of JobQueue.h.
- *
- * Data Flow: suspend / post / resume cycle
- * =========================================
- *
- *         coroutine body                 CoroTaskRunner              JobQueue
- *         --------------                 --------------              --------
- *              |
- *    co_await runner->suspend()
- *              |
- *              +--- await_suspend ------> onSuspend()
- *              |                            ++nSuspend_ ------------> nSuspend_
- *              |                          [coroutine is now suspended]
- *              |
- *              .    (externally or by JobQueueAwaiter)
- *              .
- *              +--- (caller calls) -----> post()
- *              |                            ++runCount_
- *              |                            addJob(resume) ----------> job enqueued
- *              |                                                        |
- *              |                                                      [worker picks up]
- *              |                                                        |
- *              +--- <----- resume() <-----------------------------------+
- *              |             --nSuspend_ ------> nSuspend_
- *              |             swap in LocalValues (lvs_)
- *              |             task_.handle().resume()
- *              |                |
- *              |    [coroutine body continues here]
- *              |                |
- *              |             swap out LocalValues
- *              |             --runCount_
- *              |             cv_.notify_all()
- *              v
- *
- * Thread Safety
- * =============
- * - mutex_     : guards task_.handle().resume() so that post()-before-suspend
- *                races cannot resume the coroutine while it is still running.
- *                (See the race condition discussion in JobQueue.h)
- * - mutex_run_ : guards runCount_ counter; used by join() to wait until
- *                all in-flight resume operations complete.
- * - jq_.m_mutex: guards nSuspend_ increments/decrements.
- *
- * Common Mistakes When Modifying This File
- * =========================================
- *
- * 1. Changing lock ordering.
- *    resume() acquires locks in this order: jq_.m_mutex -> mutex_ -> mutex_run_.
- *    post() acquires only mutex_run_. Any new code path that touches these
- *    mutexes must follow the same order to avoid deadlocks.
- *
- * 2. Removing the shared_from_this() capture in post().
- *    The lambda passed to addJob captures [this, sp = shared_from_this()].
- *    If you remove sp, 'this' can be destroyed before the job runs,
- *    causing use-after-free. The sp capture is load-bearing.
- *
- * 3. Forgetting to decrement nSuspend_ on a new code path.
- *    Every ++nSuspend_ must have a matching --nSuspend_. If you add a new
- *    suspension path (e.g. a new awaiter) and forget to decrement on resume
- *    or on failure, JobQueue::stop() will hang.
- *
- * 4. Calling task_.handle().resume() without holding mutex_.
- *    This allows a race where the coroutine runs on two threads
- *    simultaneously. Always hold mutex_ around resume().
- *
- * 5. Swapping LocalValues outside of the mutex_ critical section.
- *    The swap-in and swap-out of LocalValues must bracket the resume()
- *    call. If you move the swap-out before the lock_guard(mutex_) is
- *    released, you break LocalValue isolation for any code that runs
- *    after the coroutine suspends but before the lock is dropped.
- */
-//
-
-namespace xrpl {
-
-/**
- * Construct a CoroTaskRunner. Sets runCount_ to 0; does not
- * create the coroutine. Call init() afterwards.
- *
- * @param jq   The JobQueue this coroutine will run on
- * @param type Job type for scheduling priority
- * @param name Human-readable name for logging
- */
-inline JobQueue::CoroTaskRunner::CoroTaskRunner(
-    create_t,
-    JobQueue& jq,
-    JobType type,
-    std::string const& name)
-    : jq_(jq), type_(type), name_(name), runCount_(0)
-{
-}
-
-/**
- * Initialize with a coroutine-returning callable.
- * Stores the callable on the heap (FuncStore) so it outlives the
- * coroutine frame. Coroutine frames store a reference to the
- * callable's implicit object parameter (the lambda). If the callable
- * is a temporary, that reference dangles after the caller returns.
- * Keeping the callable alive here ensures the coroutine's captures
- * remain valid.
- *
- * @param f Callable: CoroTask<void>(shared_ptr<CoroTaskRunner>)
- */
-template <class F>
-void
-JobQueue::CoroTaskRunner::init(F&& f)
-{
-    using Fn = std::decay_t<F>;
-    auto store = std::make_unique<FuncStore<Fn>>(std::forward<F>(f));
-    task_ = store->func(shared_from_this());
-    storedFunc_ = std::move(store);
-}
-
-/**
- * Destructor. Waits for any in-flight resume() to complete, then
- * asserts (debug) that the coroutine has finished or
- * expectEarlyExit() was called.
- *
- * The join() call is necessary because with async dispatch the
- * coroutine runs on a worker thread. The gate signal (which wakes
- * the test thread) can arrive before resume() has set finished_.
- * join() synchronizes via mutex_run_, establishing a happens-before
- * edge: finished_ = true → unlock(mutex_run_) in resume() →
- * lock(mutex_run_) in join() → read finished_.
- */
-inline JobQueue::CoroTaskRunner::~CoroTaskRunner()
-{
-#ifndef NDEBUG
-    join();
-    XRPL_ASSERT(finished_, "xrpl::JobQueue::CoroTaskRunner::~CoroTaskRunner : is finished");
-#endif
-}
-
-/**
- * Increment the JobQueue's suspended-coroutine count (nSuspend_).
- */
-inline void
-JobQueue::CoroTaskRunner::onSuspend()
-{
-    std::lock_guard lock(jq_.m_mutex);
-    ++jq_.nSuspend_;
-}
-
-/**
- * Decrement nSuspend_ without resuming.
- */
-inline void
-JobQueue::CoroTaskRunner::onUndoSuspend()
-{
-    std::lock_guard lock(jq_.m_mutex);
-    --jq_.nSuspend_;
-}
-
-/**
- * Return a SuspendAwaiter whose await_suspend() increments nSuspend_
- * before the coroutine actually suspends. The caller must later call
- * post() or resume() to continue execution.
- *
- * @return Awaiter for use with `co_await runner->suspend()`
- */
-inline auto
-JobQueue::CoroTaskRunner::suspend()
-{
-    /**
-     * Custom awaiter for suspend(). Always suspends (await_ready
-     * returns false) and increments nSuspend_ in await_suspend().
-     */
-    struct SuspendAwaiter
-    {
-        CoroTaskRunner& runner_;  // The runner that owns this coroutine.
-
-        /**
-         * Always returns false so the coroutine suspends.
-         */
-        bool
-        await_ready() const noexcept
-        {
-            return false;
-        }
-
-        /**
-         * Called when the coroutine suspends. Increments nSuspend_
-         * so the JobQueue knows a coroutine is waiting.
-         */
-        void
-        await_suspend(std::coroutine_handle<>) const
-        {
-            runner_.onSuspend();
-        }
-
-        void
-        await_resume() const noexcept
-        {
-        }
-    };
-    return SuspendAwaiter{*this};
-}
-
-/**
- * Schedule coroutine resumption as a job on the JobQueue.
- * A shared_ptr capture (sp) prevents this CoroTaskRunner from being
- * destroyed while the job is queued but not yet executed.
- *
- * @return false if the JobQueue rejected the job (shutting down)
- */
-inline bool
-JobQueue::CoroTaskRunner::post()
-{
-    {
-        std::lock_guard lk(mutex_run_);
-        ++runCount_;
-    }
-
-    // sp prevents 'this' from being destroyed while the job is pending
-    if (jq_.addJob(type_, name_, [this, sp = shared_from_this()]() { resume(); }))
-    {
-        return true;
-    }
-
-    // The coroutine will not run. Undo the runCount_ increment.
-    std::lock_guard lk(mutex_run_);
-    --runCount_;
-    cv_.notify_all();
-    return false;
-}
-
-/**
- * Resume the coroutine on the current thread.
- *
- * Steps:
- *   1. Decrement nSuspend_ (under jq_.m_mutex)
- *   2. Swap in this coroutine's LocalValues for thread-local isolation
- *   3. Resume the coroutine handle (under mutex_)
- *   4. Swap out LocalValues, restoring the thread's previous state
- *   5. Decrement runCount_ and notify join() waiters
- *
- * Note: runCount_ is NOT incremented here — post() already did that.
- * This ensures join() stays blocked for the entire post→resume lifetime.
- */
-inline void
-JobQueue::CoroTaskRunner::resume()
-{
-    {
-        std::lock_guard lock(jq_.m_mutex);
-        --jq_.nSuspend_;
-    }
-    auto saved = detail::getLocalValues().release();
-    detail::getLocalValues().reset(&lvs_);
-    std::lock_guard lock(mutex_);
-    XRPL_ASSERT(!task_.done(), "xrpl::JobQueue::CoroTaskRunner::resume : task is not done");
-    task_.handle().resume();
-    detail::getLocalValues().release();
-    detail::getLocalValues().reset(saved);
-    if (task_.done())
-    {
-#ifndef NDEBUG
-        finished_ = true;
-#endif
-        // Break the shared_ptr cycle: frame -> shared_ptr<runner> -> this.
-        // Use std::move (not task_ = {}) so task_.handle_ is null BEFORE the
-        // frame is destroyed. operator= would destroy the frame while handle_
-        // still holds the old value -- a re-entrancy hazard on GCC-12 if
-        // frame destruction triggers runner cleanup.
-        [[maybe_unused]] auto completed = std::move(task_);
-    }
-    std::lock_guard lk(mutex_run_);
-    --runCount_;
-    cv_.notify_all();
-}
-
-/**
- * @return true if the coroutine has not yet run to completion
- */
-inline bool
-JobQueue::CoroTaskRunner::runnable() const
-{
-    // After normal completion, task_ is reset to break the shared_ptr cycle
-    // (handle_ becomes null). A null handle means the coroutine is done.
-    return task_.handle() && !task_.done();
-}
-
-/**
- * Handle early termination when the coroutine never ran (e.g. JobQueue
- * is stopping). Decrements nSuspend_ and destroys the coroutine frame
- * to break the shared_ptr cycle: frame -> lambda -> runner -> frame.
- */
-inline void
-JobQueue::CoroTaskRunner::expectEarlyExit()
-{
-#ifndef NDEBUG
-    if (!finished_)
-#endif
-    {
-        std::lock_guard lock(jq_.m_mutex);
-        --jq_.nSuspend_;
-#ifndef NDEBUG
-        finished_ = true;
-#endif
-    }
-    // Break the shared_ptr cycle: frame -> shared_ptr<runner> -> this.
-    // The coroutine is at initial_suspend and never ran user code, so
-    // destroying it is safe. Use std::move (not task_ = {}) so
-    // task_.handle_ is null before the frame is destroyed.
-    {
-        [[maybe_unused]] auto completed = std::move(task_);
-    }
-    storedFunc_.reset();
-}
-
-/**
- * Block until all pending/active resume operations complete.
- * Uses cv_ + mutex_run_ to wait until runCount_ reaches 0.
- */
-inline void
-JobQueue::CoroTaskRunner::join()
-{
-    std::unique_lock<std::mutex> lk(mutex_run_);
-    cv_.wait(lk, [this]() { return runCount_ == 0; });
-}
-
-}  // namespace xrpl

include/xrpl/core/JobQueue.h

@@ -2,7 +2,6 @@
 
 #include <xrpl/basics/LocalValue.h>
 #include <xrpl/core/ClosureCounter.h>
-#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/JobTypeData.h>
 #include <xrpl/core/JobTypes.h>
 #include <xrpl/core/detail/Workers.h>
@@ -10,7 +9,6 @@
 
 #include <boost/coroutine/all.hpp>
 
-#include <coroutine>
 #include <set>
 
 namespace xrpl {
@@ -121,395 +119,6 @@ public:
         join();
     };
 
-    /** C++20 coroutine lifecycle manager. Replaces Coro for new code.
-     *
-     *  Class / Inheritance / Dependency Diagram
-     *  =========================================
-     *
-     *  std::enable_shared_from_this<CoroTaskRunner>
-     *            ^
-     *            | (public inheritance)
-     *            |
-     *  CoroTaskRunner
-     *  +---------------------------------------------------+
-     *  | - lvs_         : detail::LocalValues               |
-     *  | - jq_          : JobQueue&                         |
-     *  | - type_        : JobType                           |
-     *  | - name_        : std::string                       |
-     *  | - runCount_    : int         (in-flight resumes)    |
-     *  | - mutex_       : std::mutex      (coroutine guard) |
-     *  | - mutex_run_   : std::mutex      (join guard)      |
-     *  | - cv_          : condition_variable                 |
-     *  | - task_        : CoroTask<void>                    |
-     *  | - storedFunc_  : unique_ptr<FuncBase> (type-erased)|
-     *  +---------------------------------------------------+
-     *  | + init(F&&)         : set up coroutine callable    |
-     *  | + onSuspend()       : ++jq_.nSuspend_              |
-     *  | + onUndoSuspend()   : --jq_.nSuspend_              |
-     *  | + suspend()         : returns SuspendAwaiter       |
-     *  | + post()            : schedule resume on JobQueue  |
-     *  | + resume()          : resume coroutine on caller   |
-     *  | + runnable()        : !task_.done()                |
-     *  | + expectEarlyExit() : teardown for failed post     |
-     *  | + join()            : block until not running       |
-     *  +---------------------------------------------------+
-     *          |                          |
-     *          | owns                     | references
-     *          v                          v
-     *    CoroTask<void>             JobQueue
-     *    (coroutine frame)          (thread pool + nSuspend_)
-     *
-     *    FuncBase / FuncStore<F>    (type-erased heap storage
-     *                                for the coroutine lambda)
-     *
-     *  Coroutine Lifecycle (Control Flow)
-     *  ===================================
-     *
-     *  Caller thread              JobQueue worker thread
-     *  -------------              ----------------------
-     *  postCoroTask(f)
-     *    |
-     *    +-- check stopping_ (reject if JQ shutting down)
-     *    +-- ++nSuspend_  (lazy start counts as suspended)
-     *    +-- make_shared<CoroTaskRunner>
-     *    +-- init(f)
-     *    |     +-- store lambda on heap (FuncStore)
-     *    |     +-- task_ = f(shared_from_this())
-     *    |           [coroutine created, suspended at initial_suspend]
-     *    +-- post()
-     *    |     +-- ++runCount_
-     *    |     +-- addJob(type_, [resume]{})
-     *    |                                    resume()
-     *    |                                      |
-     *    |                                      +-- --nSuspend_
-     *    |                                      +-- swap in LocalValues
-     *    |                                      +-- task_.handle().resume()
-     *    |                                      |     [coroutine body runs]
-     *    |                                      |     ...
-     *    |                                      |     co_await suspend()
-     *    |                                      |       +-- ++nSuspend_
-     *    |                                      |       [coroutine suspends]
-     *    |                                      +-- swap out LocalValues
-     *    |                                      +-- --runCount_
-     *    |                                      +-- cv_.notify_all()
-     *    |
-     *  post()  <-- called externally or by JobQueueAwaiter
-     *    +-- ++runCount_
-     *    +-- addJob(type_, [resume]{})
-     *                                         resume()
-     *                                           |
-     *                                           +-- [coroutine body continues]
-     *                                           +-- co_return
-     *                                           +-- --runCount_
-     *                                           +-- cv_.notify_all()
-     *  join()
-     *    +-- cv_.wait([]{runCount_ == 0})
-     *    +-- [done]
-     *
-     *  Usage Examples
-     *  ==============
-     *
-     *  1. Fire-and-forget coroutine (most common pattern):
-     *
-     *      jq.postCoroTask(jtCLIENT, "MyWork",
-     *          [](auto runner) -> CoroTask<void> {
-     *              doSomeWork();
-     *              co_await runner->suspend();  // yield to other jobs
-     *              doMoreWork();
-     *              co_return;
-     *          });
-     *
-     *  2. Manually controlling suspend / resume (external trigger):
-     *
-     *      auto runner = jq.postCoroTask(jtCLIENT, "ExtTrigger",
-     *          [&result](auto runner) -> CoroTask<void> {
-     *              startAsyncOperation(callback);
-     *              co_await runner->suspend();
-     *              // callback called runner->post() to get here
-     *              result = collectResult();
-     *              co_return;
-     *          });
-     *      // ... later, from the callback:
-     *      runner->post();   // reschedule the coroutine on the JobQueue
-     *
-     *  3. Using JobQueueAwaiter for automatic suspend + repost:
-     *
-     *      jq.postCoroTask(jtCLIENT, "AutoRepost",
-     *          [](auto runner) -> CoroTask<void> {
-     *              step1();
-     *              co_await JobQueueAwaiter{runner};  // yield + auto-repost
-     *              step2();
-     *              co_await JobQueueAwaiter{runner};
-     *              step3();
-     *              co_return;
-     *          });
-     *
-     *  4. Checking shutdown after co_await (cooperative cancellation):
-     *
-     *      jq.postCoroTask(jtCLIENT, "Cancellable",
-     *          [&jq](auto runner) -> CoroTask<void> {
-     *              while (moreWork()) {
-     *                  co_await JobQueueAwaiter{runner};
-     *                  if (jq.isStopping())
-     *                      co_return;  // bail out cleanly
-     *                  processNextItem();
-     *              }
-     *              co_return;
-     *          });
-     *
-     *  Caveats / Pitfalls
-     *  ==================
-     *
-     *  BUG-RISK: Calling suspend() without a matching post()/resume().
-     *    After co_await runner->suspend(), the coroutine is parked and
-     *    nSuspend_ is incremented.  If nothing ever calls post() or
-     *    resume(), the coroutine is leaked and JobQueue::stop() will
-     *    hang forever waiting for nSuspend_ to reach zero.
-     *
-     *  BUG-RISK: Calling post() on an already-running coroutine.
-     *    post() schedules a resume() job.  If the coroutine has not
-     *    actually suspended yet (no co_await executed), the resume job
-     *    will try to call handle().resume() while the coroutine is still
-     *    running on another thread.  This is UB.  The mutex_ prevents
-     *    data corruption but the logic is wrong — always co_await
-     *    suspend() before calling post().  (The test testIncorrectOrder
-     *    shows this works only because mutex_ serializes the calls.)
-     *
-     *  BUG-RISK: Dropping the shared_ptr<CoroTaskRunner> before join().
-     *    The CoroTaskRunner destructor asserts (!finished_ is false).
-     *    If you let the last shared_ptr die while the coroutine is still
-     *    running or suspended, you get an assertion failure in debug and
-     *    UB in release.  Always call join() or expectEarlyExit() first.
-     *
-     *  BUG-RISK: Lambda captures outliving the coroutine frame.
-     *    The lambda passed to postCoroTask is heap-allocated (FuncStore)
-     *    to prevent dangling.  But objects captured by pointer still need
-     *    their own lifetime management.  If you capture a raw pointer to
-     *    a stack variable, and the stack frame exits before the coroutine
-     *    finishes, the pointer dangles.  Use shared_ptr or ensure the
-     *    pointed-to object outlives the coroutine.
-     *
-     *  BUG-RISK: Forgetting co_return in a void coroutine.
-     *    If the coroutine body falls off the end without co_return,
-     *    the compiler may silently treat it as co_return (per standard),
-     *    but some compilers warn.  Always write explicit co_return.
-     *
-     *  LIMITATION: CoroTaskRunner only supports CoroTask<void>.
-     *    The task_ member is CoroTask<void>.  To return values from
-     *    the top-level coroutine, write through a captured pointer
-     *    (as the tests demonstrate), or co_await inner CoroTask<T>
-     *    coroutines that return values.
-     *
-     *  LIMITATION: One coroutine per CoroTaskRunner.
-     *    init() must be called exactly once.  You cannot reuse a
-     *    CoroTaskRunner to run a second coroutine.  Create a new one
-     *    via postCoroTask() instead.
-     *
-     *  LIMITATION: No timeout on join().
-     *    join() blocks indefinitely.  If the coroutine is suspended
-     *    and never posted, join() will deadlock.  Use timed waits
-     *    on the gate pattern (condition_variable + wait_for) in tests.
-     */
-    class CoroTaskRunner : public std::enable_shared_from_this<CoroTaskRunner>
-    {
-    private:
-        // Per-coroutine thread-local storage. Swapped in before resume()
-        // and swapped out after, so each coroutine sees its own LocalValue
-        // state regardless of which worker thread executes it.
-        detail::LocalValues lvs_;
-
-        // Back-reference to the owning JobQueue. Used to post jobs,
-        // increment/decrement nSuspend_, and acquire jq_.m_mutex.
-        JobQueue& jq_;
-
-        // Job type passed to addJob() when posting this coroutine.
-        JobType type_;
-
-        // Human-readable name for this coroutine job (for logging).
-        std::string name_;
-
-        // Number of in-flight resume operations (pending + active).
-        // Incremented by post(), decremented when resume() finishes.
-        // Guarded by mutex_run_. join() blocks until this reaches 0.
-        //
-        // A counter (not a bool) is needed because post() can be called
-        // from within the coroutine body (e.g. via JobQueueAwaiter),
-        // enqueuing a second resume while the first is still running.
-        // A bool would be clobbered: R2.post() sets true, then R1's
-        // cleanup sets false — losing the fact that R2 is still pending.
-        int runCount_;
-
-        // Guards task_.handle().resume() to prevent the coroutine from
-        // running on two threads simultaneously. Handles the race where
-        // post() enqueues a resume before the coroutine has actually
-        // suspended (post-before-suspend pattern).
-        std::mutex mutex_;
-
-        // Guards runCount_. Used with cv_ for join() to wait
-        // until all pending/active resume operations complete.
-        std::mutex mutex_run_;
-
-        // Notified when runCount_ reaches zero, allowing
-        // join() waiters to wake up.
-        std::condition_variable cv_;
-
-        // The coroutine handle wrapper. Owns the coroutine frame.
-        // Set by init(), reset to empty by expectEarlyExit() on
-        // early termination.
-        CoroTask<void> task_;
-
-        /**
-         * Type-erased base for heap-stored callables.
-         * Prevents the coroutine lambda from being destroyed before
-         * the coroutine frame is done with it.
-         *
-         * @see FuncStore
-         */
-        struct FuncBase
-        {
-            virtual ~FuncBase() = default;
-        };
-
-        /**
-         * Concrete type-erased storage for a callable of type F.
-         * The coroutine frame stores a reference to the lambda's implicit
-         * object parameter. If the lambda is a temporary, that reference
-         * dangles after the call returns. FuncStore keeps it alive on
-         * the heap for the lifetime of the CoroTaskRunner.
-         */
-        template <class F>
-        struct FuncStore : FuncBase
-        {
-            F func;  // The stored callable (coroutine lambda).
-            explicit FuncStore(F&& f) : func(std::move(f))
-            {
-            }
-        };
-
-        // Heap-allocated callable storage. Set by init(), ensures the
-        // lambda outlives the coroutine frame that references it.
-        std::unique_ptr<FuncBase> storedFunc_;
-
-#ifndef NDEBUG
-        // Debug-only flag. True once the coroutine has completed or
-        // expectEarlyExit() was called. Asserted in the destructor
-        // to catch leaked runners.
-        bool finished_ = false;
-#endif
-
-    public:
-        /**
-         * Tag type for private construction. Prevents external code
-         * from constructing CoroTaskRunner directly. Use postCoroTask().
-         */
-        struct create_t
-        {
-            explicit create_t() = default;
-        };
-
-        /**
-         * Construct a CoroTaskRunner. Private by convention (create_t tag).
-         *
-         * @param jq   The JobQueue this coroutine will run on
-         * @param type Job type for scheduling priority
-         * @param name Human-readable name for logging
-         */
-        CoroTaskRunner(create_t, JobQueue&, JobType, std::string const&);
-
-        CoroTaskRunner(CoroTaskRunner const&) = delete;
-        CoroTaskRunner&
-        operator=(CoroTaskRunner const&) = delete;
-
-        /**
-         * Destructor. Asserts (debug) that the coroutine has finished
-         * or expectEarlyExit() was called.
-         */
-        ~CoroTaskRunner();
-
-        /**
-         * Initialize with a coroutine-returning callable.
-         * Must be called exactly once, after the object is managed by
-         * shared_ptr (because init uses shared_from_this internally).
-         * This is handled automatically by postCoroTask().
-         *
-         * @param f Callable: CoroTask<void>(shared_ptr<CoroTaskRunner>)
-         */
-        template <class F>
-        void
-        init(F&& f);
-
-        /**
-         * Increment the JobQueue's suspended-coroutine count (nSuspend_).
-         * Called when the coroutine is about to suspend. Every call
-         * must be balanced by a corresponding decrement (via resume()
-         * or onUndoSuspend()), or JobQueue::stop() will hang.
-         */
-        void
-        onSuspend();
-
-        /**
-         * Decrement nSuspend_ without resuming.
-         * Used to undo onSuspend() when a scheduled post() fails
-         * (e.g. JobQueue is stopping).
-         */
-        void
-        onUndoSuspend();
-
-        /**
-         * Suspend the coroutine.
-         * The awaiter's await_suspend() increments nSuspend_ before the
-         * coroutine actually suspends. The caller must later call post()
-         * or resume() to continue execution.
-         *
-         * @return An awaiter for use with `co_await runner->suspend()`
-         */
-        auto
-        suspend();
-
-        /**
-         * Schedule coroutine resumption as a job on the JobQueue.
-         * Captures shared_from_this() to prevent this runner from being
-         * destroyed while the job is queued.
-         *
-         * @return true if the job was accepted; false if the JobQueue
-         *         is stopping (caller must handle cleanup)
-         */
-        bool
-        post();
-
-        /**
-         * Resume the coroutine on the current thread.
-         * Decrements nSuspend_, swaps in LocalValues, resumes the
-         * coroutine handle, swaps out LocalValues, and notifies join()
-         * waiters. Lock ordering: mutex_run_ -> jq_.m_mutex -> mutex_.
-         */
-        void
-        resume();
-
-        /**
-         * @return true if the coroutine has not yet run to completion
-         */
-        bool
-        runnable() const;
-
-        /**
-         * Handle early termination when the coroutine never ran.
-         * Decrements nSuspend_ and destroys the coroutine frame to
-         * break the shared_ptr cycle (frame -> lambda -> runner -> frame).
-         * Called by postCoroTask() when post() fails.
-         */
-        void
-        expectEarlyExit();
-
-        /**
-         * Block until all pending/active resume operations complete.
-         * Uses cv_ + mutex_run_ to wait until runCount_ reaches 0.
-         * Warning: deadlocks if the coroutine is suspended and never posted.
-         */
-        void
-        join();
-    };
-
     using JobFunction = std::function<void()>;
 
     JobQueue(
@@ -556,19 +165,6 @@ public:
     std::shared_ptr<Coro>
     postCoro(JobType t, std::string const& name, F&& f);
 
-    /** Creates a C++20 coroutine and adds a job to the queue to run it.
-
-        @param t The type of job.
-        @param name Name of the job.
-        @param f Callable with signature
-            CoroTask<void>(std::shared_ptr<CoroTaskRunner>).
-
-        @return shared_ptr to posted CoroTaskRunner. nullptr if not successful.
-    */
-    template <class F>
-    std::shared_ptr<CoroTaskRunner>
-    postCoroTask(JobType t, std::string const& name, F&& f);
-
     /** Jobs waiting at this priority.
      */
     int
@@ -783,7 +379,6 @@ private:
 }  // namespace xrpl
 
 #include <xrpl/core/Coro.ipp>
-#include <xrpl/core/CoroTaskRunner.ipp>
 
 namespace xrpl {
 
@@ -806,69 +401,4 @@ JobQueue::postCoro(JobType t, std::string const& name, F&& f)
     return coro;
 }
 
-// postCoroTask — entry point for launching a C++20 coroutine on the JobQueue.
-//
-// Control Flow
-// ============
-//
-//   postCoroTask(t, name, f)
-//     |
-//     +-- 1. Check stopping_ — reject if JQ shutting down
-//     |
-//     +-- 2. ++nSuspend_   (mirrors Boost Coro ctor's implicit yield)
-//     |        The coroutine is "suspended" from the JobQueue's perspective
-//     |        even though it hasn't run yet — this keeps the JQ shutdown
-//     |        logic correct (it waits for nSuspend_ to reach 0).
-//     |
-//     +-- 3. Create CoroTaskRunner (shared_ptr, ref-counted)
-//     |
-//     +-- 4. runner->init(f)
-//     |        +-- Heap-allocate the lambda (FuncStore) to prevent
-//     |        |   dangling captures in the coroutine frame
-//     |        +-- task_ = f(shared_from_this())
-//     |              [coroutine created but NOT started — lazy initial_suspend]
-//     |
-//     +-- 5. runner->post()
-//     |        +-- addJob(type_, [resume]{})   → resume on worker thread
-//     |        +-- failure (JQ stopping):
-//     |              +-- runner->expectEarlyExit()
-//     |              |     --nSuspend_, destroy coroutine frame
-//     |              +-- return nullptr
-//
-// Why async post() instead of synchronous resume()?
-// ==================================================
-// The initial dispatch MUST use async post() so the coroutine body runs on
-// a JobQueue worker thread, not the caller's thread.  resume() swaps the
-// caller's thread-local LocalValues with the coroutine's private copy.
-// If the coroutine mutates LocalValues (e.g. thread_specific_storage test),
-// those mutations bleed back into the caller's thread-local state after the
-// swap-out, corrupting subsequent tests that share the same thread pool.
-// Async post() avoids this by running the coroutine on a worker thread whose
-// LocalValues are managed by the thread pool, not by the caller.
-//
-template <class F>
-std::shared_ptr<JobQueue::CoroTaskRunner>
-JobQueue::postCoroTask(JobType t, std::string const& name, F&& f)
-{
-    // Reject if the JQ is shutting down — matches addJob()'s stopping_ check.
-    // Must check before incrementing nSuspend_ to avoid leaving an orphan
-    // count that would cause stop() to hang.
-    if (stopping_)
-        return nullptr;
-
-    {
-        std::lock_guard lock(m_mutex);
-        ++nSuspend_;
-    }
-
-    auto runner = std::make_shared<CoroTaskRunner>(CoroTaskRunner::create_t{}, *this, t, name);
-    runner->init(std::forward<F>(f));
-    if (!runner->post())
-    {
-        runner->expectEarlyExit();
-        runner.reset();
-    }
-    return runner;
-}
-
 }  // namespace xrpl

include/xrpl/core/JobQueueAwaiter.h

@@ -1,174 +0,0 @@
-#pragma once
-
-#include <xrpl/core/JobQueue.h>
-
-#include <coroutine>
-#include <memory>
-
-namespace xrpl {
-
-/**
- * Awaiter that suspends and immediately reschedules on the JobQueue.
- * Equivalent to calling yield() followed by post() in the old Coro API.
- *
- * Usage:
- *     co_await JobQueueAwaiter{runner};
- *
- * What it waits for: The coroutine is re-queued as a job and resumes
- * when a worker thread picks it up.
- *
- * Which thread resumes: A JobQueue worker thread.
- *
- * What await_resume() returns: void.
- *
- * Dependency Diagram
- * ==================
- *
- *   JobQueueAwaiter
- *   +----------------------------------------------+
- *   | + runner : shared_ptr<CoroTaskRunner>         |
- *   +----------------------------------------------+
- *   | + await_ready()   -> false (always suspend)   |
- *   | + await_suspend() -> bool (suspend or cancel) |
- *   | + await_resume()  -> void                     |
- *   +----------------------------------------------+
- *          |                         |
- *          | uses                    | uses
- *          v                         v
- *   CoroTaskRunner              JobQueue
- *   .onSuspend()                (via runner->post() -> addJob)
- *   .onUndoSuspend()
- *   .post()
- *
- * Control Flow (await_suspend)
- * ============================
- *
- *   co_await JobQueueAwaiter{runner}
- *     |
- *     +-- await_ready() -> false
- *     +-- await_suspend(handle)
- *           |
- *           +-- runner->onSuspend()     // ++nSuspend_
- *           +-- runner->post()          // addJob to JobQueue
- *           |     |
- *           |     +-- success? return true   // coroutine stays suspended
- *           |     |                          // worker thread will call resume()
- *           |     +-- failure? (JQ stopping)
- *           |           +-- runner->onUndoSuspend()  // --nSuspend_
- *           |           +-- return false   // coroutine continues immediately
- *           |                              // so it can clean up and co_return
- *
- * Usage Examples
- * ==============
- *
- * 1. Yield and auto-repost (most common -- replaces yield() + post()):
- *
- *     CoroTask<void> handler(auto runner) {
- *         doPartA();
- *         co_await JobQueueAwaiter{runner};  // yield + repost
- *         doPartB();                         // runs on a worker thread
- *         co_return;
- *     }
- *
- * 2. Multiple yield points in a loop:
- *
- *     CoroTask<void> batchProcessor(auto runner) {
- *         for (auto& item : items) {
- *             process(item);
- *             co_await JobQueueAwaiter{runner};  // let other jobs run
- *         }
- *         co_return;
- *     }
- *
- * 3. Graceful shutdown -- checking after resume:
- *
- *     CoroTask<void> longTask(auto runner, JobQueue& jq) {
- *         while (hasWork()) {
- *             co_await JobQueueAwaiter{runner};
- *             // If JQ is stopping, await_suspend returns false and
- *             // the coroutine continues immediately without re-queuing.
- *             // Always check isStopping() to decide whether to proceed:
- *             if (jq.isStopping())
- *                 co_return;
- *             doNextChunk();
- *         }
- *         co_return;
- *     }
- *
- * Caveats / Pitfalls
- * ==================
- *
- * BUG-RISK: Using a stale or null runner.
- *   The runner shared_ptr must be valid and point to the CoroTaskRunner
- *   that owns the coroutine currently executing. Passing a runner from
- *   a different coroutine, or a default-constructed shared_ptr, is UB.
- *
- * BUG-RISK: Assuming resume happens on the same thread.
- *   After co_await JobQueueAwaiter, the coroutine resumes on whatever
- *   worker thread picks up the job. Do not rely on thread-local state
- *   unless it is managed through LocalValue (which CoroTaskRunner
- *   automatically swaps in/out).
- *
- * BUG-RISK: Ignoring the shutdown path.
- *   When the JobQueue is stopping, post() fails and await_suspend()
- *   returns false (coroutine does NOT actually suspend). The coroutine
- *   body continues immediately on the same thread. If your code after
- *   co_await assumes it was re-queued and is running on a worker thread,
- *   that assumption breaks during shutdown. Always handle the "JQ is
- *   stopping" case, either by checking jq.isStopping() or by letting
- *   the coroutine fall through to co_return naturally.
- *
- * DIFFERENCE from runner->suspend() + runner->post():
- *   JobQueueAwaiter combines both in one atomic operation. With the
- *   manual suspend()/post() pattern, there is a window between the
- *   two calls where an external event could race. JobQueueAwaiter
- *   removes that window -- onSuspend() and post() happen within the
- *   same await_suspend() call while the coroutine is guaranteed to
- *   be suspended. Prefer JobQueueAwaiter unless you need an external
- *   party to decide *when* to call post().
- */
-struct JobQueueAwaiter
-{
-    // The CoroTaskRunner that owns the currently executing coroutine.
-    std::shared_ptr<JobQueue::CoroTaskRunner> runner;
-
-    /**
-     * Always returns false so the coroutine suspends.
-     */
-    bool
-    await_ready() const noexcept
-    {
-        return false;
-    }
-
-    /**
-     * Increment nSuspend (equivalent to yield()) and schedule resume
-     * on the JobQueue (equivalent to post()). If the JobQueue is
-     * stopping, undoes the suspend count and returns false so the
-     * coroutine continues immediately and can clean up.
-     *
-     * @return true if coroutine should stay suspended (job posted);
-     *         false if coroutine should continue (JQ stopping)
-     */
-    bool
-    await_suspend(std::coroutine_handle<>)
-    {
-        runner->onSuspend();
-        if (!runner->post())
-        {
-            // JobQueue is stopping. Undo the suspend count and
-            // don't actually suspend — the coroutine continues
-            // immediately so it can clean up and co_return.
-            runner->onUndoSuspend();
-            return false;
-        }
-        return true;
-    }
-
-    void
-    await_resume() const noexcept
-    {
-    }
-};
-
-}  // namespace xrpl

include/xrpl/nodestore/Backend.h

@@ -77,16 +77,16 @@ public:
         If the object is not found or an error is encountered, the
         result will indicate the condition.
         @note This will be called concurrently.
-        @param hash The hash of the object.
+        @param key A pointer to the key data.
         @param pObject [out] The created object if successful.
         @return The result of the operation.
     */
     virtual Status
-    fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pObject) = 0;
+    fetch(void const* key, std::shared_ptr<NodeObject>* pObject) = 0;
 
     /** Fetch a batch synchronously. */
     virtual std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
-    fetchBatch(std::vector<uint256> const& hashes) = 0;
+    fetchBatch(std::vector<uint256 const*> const& hashes) = 0;
 
     /** Store a single object.
         Depending on the implementation this may happen immediately

src/libxrpl/nodestore/DatabaseNodeImp.cpp

@@ -33,7 +33,7 @@ DatabaseNodeImp::fetchNodeObject(
 
     try
     {
-        status = backend_->fetch(hash, &nodeObject);
+        status = backend_->fetch(hash.data(), &nodeObject);
     }
     catch (std::exception const& e)
     {
@@ -68,10 +68,18 @@ DatabaseNodeImp::fetchBatch(std::vector<uint256> const& hashes)
     using namespace std::chrono;
     auto const before = steady_clock::now();
 
+    std::vector<uint256 const*> batch{};
+    batch.reserve(hashes.size());
+    for (size_t i = 0; i < hashes.size(); ++i)
+    {
+        auto const& hash = hashes[i];
+        batch.push_back(&hash);
+    }
+
     // Get the node objects that match the hashes from the backend. To protect
     // against the backends returning fewer or more results than expected, the
     // container is resized to the number of hashes.
-    auto results = backend_->fetchBatch(hashes).first;
+    auto results = backend_->fetchBatch(batch).first;
     XRPL_ASSERT(
         results.size() == hashes.size() || results.empty(),
         "number of output objects either matches number of input hashes or is empty");

src/libxrpl/nodestore/DatabaseRotatingImp.cpp

@@ -105,7 +105,7 @@ DatabaseRotatingImp::fetchNodeObject(
         std::shared_ptr<NodeObject> nodeObject;
         try
         {
-            status = backend->fetch(hash, &nodeObject);
+            status = backend->fetch(hash.data(), &nodeObject);
         }
         catch (std::exception const& e)
         {

src/libxrpl/nodestore/backend/MemoryFactory.cpp

@@ -116,9 +116,10 @@ public:
     //--------------------------------------------------------------------------
 
     Status
-    fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pObject) override
+    fetch(void const* key, std::shared_ptr<NodeObject>* pObject) override
     {
         XRPL_ASSERT(db_, "xrpl::NodeStore::MemoryBackend::fetch : non-null database");
+        uint256 const hash(uint256::fromVoid(key));
 
         std::lock_guard _(db_->mutex);
 
@@ -133,14 +134,14 @@ public:
     }
 
     std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
-    fetchBatch(std::vector<uint256> const& hashes) override
+    fetchBatch(std::vector<uint256 const*> const& hashes) override
     {
         std::vector<std::shared_ptr<NodeObject>> results;
         results.reserve(hashes.size());
         for (auto const& h : hashes)
         {
             std::shared_ptr<NodeObject> nObj;
-            Status status = fetch(h, &nObj);
+            Status status = fetch(h->begin(), &nObj);
             if (status != ok)
                 results.push_back({});
             else

src/libxrpl/nodestore/backend/NuDBFactory.cpp

@@ -179,17 +179,17 @@ public:
     }
 
     Status
-    fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pno) override
+    fetch(void const* key, std::shared_ptr<NodeObject>* pno) override
     {
         Status status;
         pno->reset();
         nudb::error_code ec;
         db_.fetch(
-            hash.data(),
-            [&hash, pno, &status](void const* data, std::size_t size) {
+            key,
+            [key, pno, &status](void const* data, std::size_t size) {
                 nudb::detail::buffer bf;
                 auto const result = nodeobject_decompress(data, size, bf);
-                DecodedBlob decoded(hash.data(), result.first, result.second);
+                DecodedBlob decoded(key, result.first, result.second);
                 if (!decoded.wasOk())
                 {
                     status = dataCorrupt;
@@ -207,14 +207,14 @@ public:
     }
 
     std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
-    fetchBatch(std::vector<uint256> const& hashes) override
+    fetchBatch(std::vector<uint256 const*> const& hashes) override
     {
         std::vector<std::shared_ptr<NodeObject>> results;
         results.reserve(hashes.size());
         for (auto const& h : hashes)
         {
             std::shared_ptr<NodeObject> nObj;
-            Status status = fetch(h, &nObj);
+            Status status = fetch(h->begin(), &nObj);
             if (status != ok)
                 results.push_back({});
             else

src/libxrpl/nodestore/backend/NullFactory.cpp

@@ -36,13 +36,13 @@ public:
     }
 
     Status
-    fetch(uint256 const&, std::shared_ptr<NodeObject>*) override
+    fetch(void const*, std::shared_ptr<NodeObject>*) override
     {
         return notFound;
     }
 
     std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
-    fetchBatch(std::vector<uint256> const& hashes) override
+    fetchBatch(std::vector<uint256 const*> const& hashes) override
     {
         return {};
     }

src/libxrpl/nodestore/backend/RocksDBFactory.cpp

@@ -244,7 +244,7 @@ public:
     //--------------------------------------------------------------------------
 
     Status
-    fetch(uint256 const& hash, std::shared_ptr<NodeObject>* pObject) override
+    fetch(void const* key, std::shared_ptr<NodeObject>* pObject) override
     {
         XRPL_ASSERT(m_db, "xrpl::NodeStore::RocksDBBackend::fetch : non-null database");
         pObject->reset();
@@ -252,7 +252,7 @@ public:
         Status status(ok);
 
         rocksdb::ReadOptions const options;
-        rocksdb::Slice const slice(std::bit_cast<char const*>(hash.data()), m_keyBytes);
+        rocksdb::Slice const slice(static_cast<char const*>(key), m_keyBytes);
 
         std::string string;
 
@@ -260,7 +260,7 @@ public:
 
         if (getStatus.ok())
         {
-            DecodedBlob decoded(hash.data(), string.data(), string.size());
+            DecodedBlob decoded(key, string.data(), string.size());
 
             if (decoded.wasOk())
             {
@@ -295,14 +295,14 @@ public:
     }
 
     std::pair<std::vector<std::shared_ptr<NodeObject>>, Status>
-    fetchBatch(std::vector<uint256> const& hashes) override
+    fetchBatch(std::vector<uint256 const*> const& hashes) override
     {
         std::vector<std::shared_ptr<NodeObject>> results;
         results.reserve(hashes.size());
         for (auto const& h : hashes)
         {
             std::shared_ptr<NodeObject> nObj;
-            Status status = fetch(h, &nObj);
+            Status status = fetch(h->begin(), &nObj);
             if (status != ok)
                 results.push_back({});
             else
@@ -332,8 +332,9 @@ public:
             EncodedBlob encoded(e);
 
             wb.Put(
-                rocksdb::Slice(std::bit_cast<char const*>(encoded.getKey()), m_keyBytes),
-                rocksdb::Slice(std::bit_cast<char const*>(encoded.getData()), encoded.getSize()));
+                rocksdb::Slice(reinterpret_cast<char const*>(encoded.getKey()), m_keyBytes),
+                rocksdb::Slice(
+                    reinterpret_cast<char const*>(encoded.getData()), encoded.getSize()));
         }
 
         rocksdb::WriteOptions const options;

src/test/app/LedgerMaster_test.cpp

@@ -2,6 +2,7 @@
 #include <test/jtx/Env.h>
 
 #include <xrpld/app/ledger/LedgerMaster.h>
+#include <xrpld/app/misc/SHAMapStore.h>
 
 namespace xrpl {
 namespace test {
@@ -95,6 +96,73 @@ class LedgerMaster_test : public beast::unit_test::suite
         }
     }
 
+    void
+    testCompleteLedgerRange(FeatureBitset features)
+    {
+        // Note that this test is intentionally very similar to
+        // SHAMapStore_test::testLedgerGaps, but has a different
+        // focus.
+
+        testcase("Complete Ledger operations");
+
+        using namespace test::jtx;
+
+        auto const deleteInterval = 8;
+
+        Env env{*this, envconfig([](auto cfg) {
+                    return online_delete(std::move(cfg), deleteInterval);
+                })};
+
+        auto const alice = Account("alice");
+        env.fund(XRP(1000), alice);
+        env.close();
+
+        auto& lm = env.app().getLedgerMaster();
+        LedgerIndex minSeq = 2;
+        LedgerIndex maxSeq = env.closed()->header().seq;
+        auto& store = env.app().getSHAMapStore();
+        store.rendezvous();
+        LedgerIndex lastRotated = store.getLastRotated();
+        BEAST_EXPECTS(maxSeq == 3, to_string(maxSeq));
+        BEAST_EXPECTS(lm.getCompleteLedgers() == "2-3", lm.getCompleteLedgers());
+        BEAST_EXPECTS(lastRotated == 3, to_string(lastRotated));
+        BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq, maxSeq) == 0);
+        BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 1, maxSeq - 1) == 0);
+        BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 1, maxSeq + 1) == 2);
+        BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 2, maxSeq - 2) == 2);
+        BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 2, maxSeq + 2) == 2);
+
+        // Close enough ledgers to rotate a few times
+        for (int i = 0; i < 24; ++i)
+        {
+            for (int t = 0; t < 3; ++t)
+            {
+                env(noop(alice));
+            }
+            env.close();
+            store.rendezvous();
+
+            ++maxSeq;
+
+            if (maxSeq == lastRotated + deleteInterval)
+            {
+                minSeq = lastRotated;
+                lastRotated = maxSeq;
+            }
+            BEAST_EXPECTS(
+                env.closed()->header().seq == maxSeq, to_string(env.closed()->header().seq));
+            BEAST_EXPECTS(store.getLastRotated() == lastRotated, to_string(store.getLastRotated()));
+            std::stringstream expectedRange;
+            expectedRange << minSeq << "-" << maxSeq;
+            BEAST_EXPECTS(lm.getCompleteLedgers() == expectedRange.str(), lm.getCompleteLedgers());
+            BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq, maxSeq) == 0);
+            BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 1, maxSeq - 1) == 0);
+            BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 1, maxSeq + 1) == 2);
+            BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 2, maxSeq - 2) == 2);
+            BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 2, maxSeq + 2) == 2);
+        }
+    }
+
 public:
     void
     run() override
@@ -108,6 +176,7 @@ public:
     testWithFeats(FeatureBitset features)
     {
         testTxnIdFromIndex(features);
+        testCompleteLedgerRange(features);
     }
 };
 

src/test/app/Path_test.cpp

@@ -8,7 +8,6 @@
 #include <xrpld/rpc/detail/Tuning.h>
 
 #include <xrpl/beast/unit_test.h>
-#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/JobQueue.h>
 #include <xrpl/json/json_reader.h>
 #include <xrpl/protocol/ApiVersion.h>
@@ -132,6 +131,7 @@ public:
              c,
              Role::USER,
              {},
+             {},
              RPC::apiVersionIfUnspecified},
             {},
             {}};
@@ -155,11 +155,11 @@ public:
 
         Json::Value result;
         gate g;
-        app.getJobQueue().postCoroTask(jtCLIENT, "RPC-Client", [&](auto) -> CoroTask<void> {
+        app.getJobQueue().postCoro(jtCLIENT, "RPC-Client", [&](auto const& coro) {
             context.params = std::move(params);
+            context.coro = coro;
             RPC::doCommand(context, result);
             g.signal();
-            co_return;
         });
 
         using namespace std::chrono_literals;
@@ -240,27 +240,28 @@ public:
              c,
              Role::USER,
              {},
+             {},
              RPC::apiVersionIfUnspecified},
             {},
             {}};
         Json::Value result;
         gate g;
         // Test RPC::Tuning::max_src_cur source currencies.
-        app.getJobQueue().postCoroTask(jtCLIENT, "RPC-Client", [&](auto) -> CoroTask<void> {
+        app.getJobQueue().postCoro(jtCLIENT, "RPC-Client", [&](auto const& coro) {
             context.params = rpf(Account("alice"), Account("bob"), RPC::Tuning::max_src_cur);
+            context.coro = coro;
             RPC::doCommand(context, result);
             g.signal();
-            co_return;
         });
         BEAST_EXPECT(g.wait_for(5s));
         BEAST_EXPECT(!result.isMember(jss::error));
 
         // Test more than RPC::Tuning::max_src_cur source currencies.
-        app.getJobQueue().postCoroTask(jtCLIENT, "RPC-Client", [&](auto) -> CoroTask<void> {
+        app.getJobQueue().postCoro(jtCLIENT, "RPC-Client", [&](auto const& coro) {
             context.params = rpf(Account("alice"), Account("bob"), RPC::Tuning::max_src_cur + 1);
+            context.coro = coro;
             RPC::doCommand(context, result);
             g.signal();
-            co_return;
         });
         BEAST_EXPECT(g.wait_for(5s));
         BEAST_EXPECT(result.isMember(jss::error));
@@ -268,22 +269,22 @@ public:
         // Test RPC::Tuning::max_auto_src_cur source currencies.
         for (auto i = 0; i < (RPC::Tuning::max_auto_src_cur - 1); ++i)
             env.trust(Account("alice")[std::to_string(i + 100)](100), "bob");
-        app.getJobQueue().postCoroTask(jtCLIENT, "RPC-Client", [&](auto) -> CoroTask<void> {
+        app.getJobQueue().postCoro(jtCLIENT, "RPC-Client", [&](auto const& coro) {
             context.params = rpf(Account("alice"), Account("bob"), 0);
+            context.coro = coro;
             RPC::doCommand(context, result);
             g.signal();
-            co_return;
         });
         BEAST_EXPECT(g.wait_for(5s));
         BEAST_EXPECT(!result.isMember(jss::error));
 
         // Test more than RPC::Tuning::max_auto_src_cur source currencies.
         env.trust(Account("alice")["AUD"](100), "bob");
-        app.getJobQueue().postCoroTask(jtCLIENT, "RPC-Client", [&](auto) -> CoroTask<void> {
+        app.getJobQueue().postCoro(jtCLIENT, "RPC-Client", [&](auto const& coro) {
             context.params = rpf(Account("alice"), Account("bob"), 0);
+            context.coro = coro;
             RPC::doCommand(context, result);
             g.signal();
-            co_return;
         });
         BEAST_EXPECT(g.wait_for(5s));
         BEAST_EXPECT(result.isMember(jss::error));

src/test/app/SHAMapStore_test.cpp

@@ -1,6 +1,7 @@
 #include <test/jtx.h>
 #include <test/jtx/envconfig.h>
 
+#include <xrpld/app/ledger/LedgerMaster.h>
 #include <xrpld/app/main/Application.h>
 #include <xrpld/app/main/NodeStoreScheduler.h>
 #include <xrpld/app/misc/SHAMapStore.h>
@@ -10,6 +11,8 @@
 #include <xrpl/nodestore/detail/DatabaseRotatingImp.h>
 #include <xrpl/protocol/jss.h>
 
+#include <thread>
+
 namespace xrpl {
 namespace test {
 
@@ -20,10 +23,8 @@ class SHAMapStore_test : public beast::unit_test::suite
     static auto
     onlineDelete(std::unique_ptr<Config> cfg)
     {
-        cfg->LEDGER_HISTORY = deleteInterval;
-        auto& section = cfg->section(ConfigSection::nodeDatabase());
-        section.set("online_delete", std::to_string(deleteInterval));
-        return cfg;
+        using namespace jtx;
+        return online_delete(std::move(cfg), deleteInterval);
     }
 
     static auto
@@ -563,6 +564,158 @@ public:
         BEAST_EXPECT(dbr->getName() == "3");
     }
 
+    void
+    testLedgerGaps()
+    {
+        // Note that this test is intentionally very similar to
+        // LedgerMaster_test::testCompleteLedgerRange, but has a different
+        // focus.
+
+        testcase("Wait for ledger gaps to fill in");
+
+        using namespace test::jtx;
+
+        Env env{*this, envconfig(onlineDelete)};
+
+        std::map<LedgerIndex, uint256> hashes;
+
+        auto failureMessage = [&](char const* label, auto expected, auto actual) {
+            std::stringstream ss;
+            ss << label << ": Expected: " << expected << ", Got: " << actual;
+            return ss.str();
+        };
+
+        auto const alice = Account("alice");
+        env.fund(XRP(1000), alice);
+        env.close();
+
+        auto& lm = env.app().getLedgerMaster();
+        LedgerIndex minSeq = 2;
+        LedgerIndex maxSeq = env.closed()->header().seq;
+        auto& store = env.app().getSHAMapStore();
+        store.rendezvous();
+        LedgerIndex lastRotated = store.getLastRotated();
+        BEAST_EXPECTS(maxSeq == 3, to_string(maxSeq));
+        BEAST_EXPECTS(lm.getCompleteLedgers() == "2-3", lm.getCompleteLedgers());
+        BEAST_EXPECTS(lastRotated == 3, to_string(lastRotated));
+        BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq, maxSeq) == 0);
+        BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 1, maxSeq - 1) == 0);
+        BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 1, maxSeq + 1) == 2);
+        BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 2, maxSeq - 2) == 2);
+        BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 2, maxSeq + 2) == 2);
+
+        // Close enough ledgers to rotate a few times
+        while (maxSeq < 20)
+        {
+            for (int t = 0; t < 3; ++t)
+            {
+                env(noop(alice));
+            }
+            env.close();
+            store.rendezvous();
+
+            ++maxSeq;
+
+            if (maxSeq + 1 == lastRotated + deleteInterval)
+            {
+                using namespace std::chrono_literals;
+
+                // The next ledger will trigger a rotation. Delete the
+                // current ledger from LedgerMaster.
+                std::this_thread::sleep_for(100ms);
+                LedgerIndex const deleteSeq = maxSeq;
+                while (!lm.haveLedger(deleteSeq))
+                {
+                    std::this_thread::sleep_for(100ms);
+                }
+                lm.clearLedger(deleteSeq);
+
+                auto expectedRange = [](auto minSeq, auto deleteSeq, auto maxSeq) {
+                    std::stringstream expectedRange;
+                    expectedRange << minSeq << "-" << (deleteSeq - 1);
+                    if (deleteSeq + 1 == maxSeq)
+                        expectedRange << "," << maxSeq;
+                    else if (deleteSeq < maxSeq)
+                        expectedRange << "," << (deleteSeq + 1) << "-" << maxSeq;
+                    return expectedRange.str();
+                };
+                BEAST_EXPECTS(
+                    lm.getCompleteLedgers() == expectedRange(minSeq, deleteSeq, maxSeq),
+                    failureMessage(
+                        "Complete ledgers",
+                        expectedRange(minSeq, deleteSeq, maxSeq),
+                        lm.getCompleteLedgers()));
+                BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq, maxSeq) == 1);
+
+                // Close another ledger, which will trigger a rotation, but the
+                // rotation will be stuck until the missing ledger is filled in.
+                env.close();
+                // DO NOT CALL rendezvous()! You'll end up with a deadlock.
+                ++maxSeq;
+
+                // Nothing has changed
+                BEAST_EXPECTS(
+                    store.getLastRotated() == lastRotated,
+                    failureMessage("lastRotated", lastRotated, store.getLastRotated()));
+                BEAST_EXPECTS(
+                    lm.getCompleteLedgers() == expectedRange(minSeq, deleteSeq, maxSeq),
+                    failureMessage(
+                        "Complete ledgers",
+                        expectedRange(minSeq, deleteSeq, maxSeq),
+                        lm.getCompleteLedgers()));
+
+                // Close 5 more ledgers, waiting one second in between to
+                // simulate the ledger making progress while online delete waits
+                // for the missing ledger to be filled in.
+                // This ensures the healthWait check has time to run and
+                // detect the gap.
+                for (int l = 0; l < 5; ++l)
+                {
+                    env.close();
+                    // DO NOT CALL rendezvous()! You'll end up with a deadlock.
+                    ++maxSeq;
+                    // Nothing has changed
+                    BEAST_EXPECTS(
+                        store.getLastRotated() == lastRotated,
+                        failureMessage("lastRotated", lastRotated, store.getLastRotated()));
+                    BEAST_EXPECTS(
+                        lm.getCompleteLedgers() == expectedRange(minSeq, deleteSeq, maxSeq),
+                        failureMessage(
+                            "Complete Ledgers",
+                            expectedRange(minSeq, deleteSeq, maxSeq),
+                            lm.getCompleteLedgers()));
+                    std::this_thread::sleep_for(1s);
+                }
+
+                // Put the missing ledger back in LedgerMaster
+                lm.setLedgerRangePresent(deleteSeq, deleteSeq);
+
+                // Wait for the rotation to finish
+                store.rendezvous();
+
+                minSeq = lastRotated;
+                lastRotated = deleteSeq + 1;
+            }
+            BEAST_EXPECT(maxSeq != lastRotated + deleteInterval);
+            BEAST_EXPECTS(
+                env.closed()->header().seq == maxSeq,
+                failureMessage("maxSeq", maxSeq, env.closed()->header().seq));
+            BEAST_EXPECTS(
+                store.getLastRotated() == lastRotated,
+                failureMessage("lastRotated", lastRotated, store.getLastRotated()));
+            std::stringstream expectedRange;
+            expectedRange << minSeq << "-" << maxSeq;
+            BEAST_EXPECTS(
+                lm.getCompleteLedgers() == expectedRange.str(),
+                failureMessage("CompleteLedgers", expectedRange.str(), lm.getCompleteLedgers()));
+            BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq, maxSeq) == 0);
+            BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 1, maxSeq - 1) == 0);
+            BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 1, maxSeq + 1) == 2);
+            BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq - 2, maxSeq - 2) == 2);
+            BEAST_EXPECT(lm.missingFromCompleteLedgerRange(minSeq + 2, maxSeq + 2) == 2);
+        }
+    }
+
     void
     run() override
     {
@@ -570,6 +723,7 @@ public:
         testAutomatic();
         testCanDelete();
         testRotate();
+        testLedgerGaps();
     }
 };
 

src/test/app/Vault_test.cpp

@@ -5340,20 +5340,20 @@ class Vault_test : public beast::unit_test::suite
             env.close();
 
             // 2. Mantissa larger than uint64 max
-            env.set_parse_failure_expected(true);
             try
             {
                 tx[sfAssetsMaximum] = "18446744073709551617e5";  // uint64 max + 1
                 env(tx, THISLINE);
-                BEAST_EXPECTS(false, "Expected parse_error for mantissa larger than uint64 max");
+                BEAST_EXPECT(false);
             }
             catch (parse_error const& e)
             {
                 using namespace std::string_literals;
                 BEAST_EXPECT(
-                    e.what() == "invalidParamsField 'tx_json.AssetsMaximum' has invalid data."s);
+                    e.what() ==
+                    "invalidParamsField 'tx_json.AssetsMaximum' has invalid "
+                    "data."s);
             }
-            env.set_parse_failure_expected(false);
         }
     }
 

src/test/core/CoroTask_test.cpp

@@ -1,537 +0,0 @@
-#include <test/jtx.h>
-
-#include <xrpl/core/JobQueue.h>
-#include <xrpl/core/JobQueueAwaiter.h>
-
-#include <chrono>
-#include <mutex>
-
-namespace xrpl {
-namespace test {
-
-/**
- * Test suite for the C++20 coroutine primitives: CoroTask, CoroTaskRunner,
- * and JobQueueAwaiter.
- *
- * Dependency Diagram
- * ==================
- *
- *   CoroTask_test
- *   +-------------------------------------------------+
- *   | + gate (inner class) : condition_variable helper |
- *   +-------------------------------------------------+
- *          |  uses
- *          v
- *   jtx::Env  -->  JobQueue::postCoroTask()
- *                       |
- *                       +-- CoroTaskRunner (suspend / post / resume)
- *                       +-- CoroTask<void> / CoroTask<T>
- *                       +-- JobQueueAwaiter
- *
- * Test Coverage Matrix
- * ====================
- *
- *   Test                      | Primitives exercised
- *   --------------------------+----------------------------------------------
- *   testVoidCompletion        | CoroTask<void> basic lifecycle
- *   testCorrectOrder          | suspend() -> join() -> post() -> complete
- *   testIncorrectOrder        | post() before suspend() (race-safe path)
- *   testJobQueueAwaiter       | JobQueueAwaiter suspend + auto-repost
- *   testThreadSpecificStorage | LocalValue isolation across coroutines
- *   testExceptionPropagation  | unhandled_exception() in promise_type
- *   testMultipleYields        | N sequential suspend/resume cycles
- *   testValueReturn           | CoroTask<T> co_return value
- *   testValueException        | CoroTask<T> exception via co_await
- *   testValueChaining         | nested CoroTask<T> -> CoroTask<T>
- *   testShutdownRejection     | postCoroTask returns nullptr when stopping
- */
-class CoroTask_test : public beast::unit_test::suite
-{
-public:
-    /**
-     * Simple one-shot gate for synchronizing between test thread
-     * and coroutine worker threads. signal() sets the flag;
-     * wait_for() blocks until signaled or timeout.
-     */
-    class gate
-    {
-    private:
-        std::condition_variable cv_;
-        std::mutex mutex_;
-        bool signaled_ = false;
-
-    public:
-        /**
-         * Block until signaled or timeout expires.
-         *
-         * @param rel_time Maximum duration to wait
-         *
-         * @return true if signaled before timeout
-         */
-        template <class Rep, class Period>
-        bool
-        wait_for(std::chrono::duration<Rep, Period> const& rel_time)
-        {
-            std::unique_lock<std::mutex> lk(mutex_);
-            auto b = cv_.wait_for(lk, rel_time, [this] { return signaled_; });
-            signaled_ = false;
-            return b;
-        }
-
-        /**
-         * Signal the gate, waking any waiting thread.
-         */
-        void
-        signal()
-        {
-            std::lock_guard lk(mutex_);
-            signaled_ = true;
-            cv_.notify_all();
-        }
-    };
-
-    // NOTE: All coroutine lambdas passed to postCoroTask use explicit
-    // pointer-by-value captures instead of [&] to work around a GCC 14
-    // bug where reference captures in coroutine lambdas are corrupted
-    // in the coroutine frame.
-
-    /**
-     * CoroTask<void> runs to completion and runner becomes non-runnable.
-     */
-    void
-    testVoidCompletion()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("void completion");
-
-        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
-            cfg->FORCE_MULTI_THREAD = true;
-            return cfg;
-        }));
-
-        gate g;
-        auto runner = env.app().getJobQueue().postCoroTask(
-            jtCLIENT, "CoroTaskTest", [gp = &g](auto) -> CoroTask<void> {
-                gp->signal();
-                co_return;
-            });
-        BEAST_EXPECT(runner);
-        BEAST_EXPECT(g.wait_for(5s));
-        runner->join();
-        BEAST_EXPECT(!runner->runnable());
-    }
-
-    /**
-     * Correct order: suspend, join, post, complete.
-     * Mirrors existing Coroutine_test::correct_order.
-     */
-    void
-    testCorrectOrder()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("correct order");
-
-        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
-            cfg->FORCE_MULTI_THREAD = true;
-            return cfg;
-        }));
-
-        gate g1, g2;
-        std::shared_ptr<JobQueue::CoroTaskRunner> r;
-        auto runner = env.app().getJobQueue().postCoroTask(
-            jtCLIENT,
-            "CoroTaskTest",
-            [rp = &r, g1p = &g1, g2p = &g2](auto runner) -> CoroTask<void> {
-                *rp = runner;
-                g1p->signal();
-                co_await runner->suspend();
-                g2p->signal();
-                co_return;
-            });
-        BEAST_EXPECT(runner);
-        BEAST_EXPECT(g1.wait_for(5s));
-        runner->join();
-        runner->post();
-        BEAST_EXPECT(g2.wait_for(5s));
-        runner->join();
-    }
-
-    /**
-     * Incorrect order: post() before suspend(). Verifies the
-     * race-safe path. Mirrors Coroutine_test::incorrect_order.
-     */
-    void
-    testIncorrectOrder()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("incorrect order");
-
-        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
-            cfg->FORCE_MULTI_THREAD = true;
-            return cfg;
-        }));
-
-        gate g;
-        env.app().getJobQueue().postCoroTask(
-            jtCLIENT, "CoroTaskTest", [gp = &g](auto runner) -> CoroTask<void> {
-                runner->post();
-                co_await runner->suspend();
-                gp->signal();
-                co_return;
-            });
-        BEAST_EXPECT(g.wait_for(5s));
-    }
-
-    /**
-     * JobQueueAwaiter suspend + auto-repost across multiple yield points.
-     */
-    void
-    testJobQueueAwaiter()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("JobQueueAwaiter");
-
-        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
-            cfg->FORCE_MULTI_THREAD = true;
-            return cfg;
-        }));
-
-        gate g;
-        int step = 0;
-        env.app().getJobQueue().postCoroTask(
-            jtCLIENT, "CoroTaskTest", [sp = &step, gp = &g](auto runner) -> CoroTask<void> {
-                *sp = 1;
-                co_await JobQueueAwaiter{runner};
-                *sp = 2;
-                co_await JobQueueAwaiter{runner};
-                *sp = 3;
-                gp->signal();
-                co_return;
-            });
-        BEAST_EXPECT(g.wait_for(5s));
-        BEAST_EXPECT(step == 3);
-    }
-
-    /**
-     * Per-coroutine LocalValue isolation. Each coroutine sees its own
-     * copy of thread-local state. Mirrors Coroutine_test::thread_specific_storage.
-     */
-    void
-    testThreadSpecificStorage()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("thread specific storage");
-        Env env(*this);
-
-        auto& jq = env.app().getJobQueue();
-
-        static int const N = 4;
-        std::array<std::shared_ptr<JobQueue::CoroTaskRunner>, N> a;
-
-        LocalValue<int> lv(-1);
-        BEAST_EXPECT(*lv == -1);
-
-        gate g;
-        jq.addJob(jtCLIENT, "LocalValTest", [&]() {
-            this->BEAST_EXPECT(*lv == -1);
-            *lv = -2;
-            this->BEAST_EXPECT(*lv == -2);
-            g.signal();
-        });
-        BEAST_EXPECT(g.wait_for(5s));
-        BEAST_EXPECT(*lv == -1);
-
-        for (int i = 0; i < N; ++i)
-        {
-            jq.postCoroTask(
-                jtCLIENT,
-                "CoroTaskTest",
-                [this, ap = &a, gp = &g, lvp = &lv, id = i](auto runner) -> CoroTask<void> {
-                    (*ap)[id] = runner;
-                    gp->signal();
-                    co_await runner->suspend();
-
-                    this->BEAST_EXPECT(**lvp == -1);
-                    **lvp = id;
-                    this->BEAST_EXPECT(**lvp == id);
-                    gp->signal();
-                    co_await runner->suspend();
-
-                    this->BEAST_EXPECT(**lvp == id);
-                    co_return;
-                });
-            BEAST_EXPECT(g.wait_for(5s));
-            a[i]->join();
-        }
-        for (auto const& r : a)
-        {
-            r->post();
-            BEAST_EXPECT(g.wait_for(5s));
-            r->join();
-        }
-        for (auto const& r : a)
-        {
-            r->post();
-            r->join();
-        }
-
-        jq.addJob(jtCLIENT, "LocalValTest", [&]() {
-            this->BEAST_EXPECT(*lv == -2);
-            g.signal();
-        });
-        BEAST_EXPECT(g.wait_for(5s));
-        BEAST_EXPECT(*lv == -1);
-    }
-
-    /**
-     * Exception thrown in coroutine body is caught by
-     * promise_type::unhandled_exception(). Coroutine completes.
-     */
-    void
-    testExceptionPropagation()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("exception propagation");
-
-        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
-            cfg->FORCE_MULTI_THREAD = true;
-            return cfg;
-        }));
-
-        gate g;
-        auto runner = env.app().getJobQueue().postCoroTask(
-            jtCLIENT, "CoroTaskTest", [gp = &g](auto) -> CoroTask<void> {
-                gp->signal();
-                throw std::runtime_error("test exception");
-                co_return;
-            });
-        BEAST_EXPECT(runner);
-        BEAST_EXPECT(g.wait_for(5s));
-        runner->join();
-        // The exception is caught by promise_type::unhandled_exception()
-        // and the coroutine is considered done
-        BEAST_EXPECT(!runner->runnable());
-    }
-
-    /**
-     * Multiple sequential suspend/resume cycles via co_await.
-     */
-    void
-    testMultipleYields()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("multiple yields");
-
-        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
-            cfg->FORCE_MULTI_THREAD = true;
-            return cfg;
-        }));
-
-        gate g;
-        int counter = 0;
-        std::shared_ptr<JobQueue::CoroTaskRunner> r;
-        auto runner = env.app().getJobQueue().postCoroTask(
-            jtCLIENT,
-            "CoroTaskTest",
-            [rp = &r, cp = &counter, gp = &g](auto runner) -> CoroTask<void> {
-                *rp = runner;
-                ++(*cp);
-                gp->signal();
-                co_await runner->suspend();
-                ++(*cp);
-                gp->signal();
-                co_await runner->suspend();
-                ++(*cp);
-                gp->signal();
-                co_return;
-            });
-        BEAST_EXPECT(runner);
-
-        BEAST_EXPECT(g.wait_for(5s));
-        BEAST_EXPECT(counter == 1);
-        runner->join();
-
-        runner->post();
-        BEAST_EXPECT(g.wait_for(5s));
-        BEAST_EXPECT(counter == 2);
-        runner->join();
-
-        runner->post();
-        BEAST_EXPECT(g.wait_for(5s));
-        BEAST_EXPECT(counter == 3);
-        runner->join();
-        BEAST_EXPECT(!runner->runnable());
-    }
-
-    /**
-     * CoroTask<T> returns a value via co_return. Outer coroutine
-     * extracts it with co_await.
-     */
-    void
-    testValueReturn()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("value return");
-
-        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
-            cfg->FORCE_MULTI_THREAD = true;
-            return cfg;
-        }));
-
-        gate g;
-        int result = 0;
-        auto runner = env.app().getJobQueue().postCoroTask(
-            jtCLIENT, "CoroTaskTest", [rp = &result, gp = &g](auto) -> CoroTask<void> {
-                auto inner = []() -> CoroTask<int> { co_return 42; };
-                *rp = co_await inner();
-                gp->signal();
-                co_return;
-            });
-        BEAST_EXPECT(runner);
-        BEAST_EXPECT(g.wait_for(5s));
-        runner->join();
-        BEAST_EXPECT(result == 42);
-        BEAST_EXPECT(!runner->runnable());
-    }
-
-    /**
-     * CoroTask<T> propagates exceptions from inner coroutines.
-     * Outer coroutine catches via try/catch around co_await.
-     */
-    void
-    testValueException()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("value exception");
-
-        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
-            cfg->FORCE_MULTI_THREAD = true;
-            return cfg;
-        }));
-
-        gate g;
-        bool caught = false;
-        auto runner = env.app().getJobQueue().postCoroTask(
-            jtCLIENT, "CoroTaskTest", [cp = &caught, gp = &g](auto) -> CoroTask<void> {
-                auto inner = []() -> CoroTask<int> {
-                    throw std::runtime_error("inner error");
-                    co_return 0;
-                };
-                try
-                {
-                    co_await inner();
-                }
-                catch (std::runtime_error const& e)
-                {
-                    *cp = true;
-                }
-                gp->signal();
-                co_return;
-            });
-        BEAST_EXPECT(runner);
-        BEAST_EXPECT(g.wait_for(5s));
-        runner->join();
-        BEAST_EXPECT(caught);
-        BEAST_EXPECT(!runner->runnable());
-    }
-
-    /**
-     * CoroTask<T> chaining. Nested value-returning coroutines
-     * compose via co_await.
-     */
-    void
-    testValueChaining()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("value chaining");
-
-        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
-            cfg->FORCE_MULTI_THREAD = true;
-            return cfg;
-        }));
-
-        gate g;
-        int result = 0;
-        auto runner = env.app().getJobQueue().postCoroTask(
-            jtCLIENT, "CoroTaskTest", [rp = &result, gp = &g](auto) -> CoroTask<void> {
-                auto add = [](int a, int b) -> CoroTask<int> { co_return a + b; };
-                auto mul = [add](int a, int b) -> CoroTask<int> {
-                    int sum = co_await add(a, b);
-                    co_return sum * 2;
-                };
-                *rp = co_await mul(3, 4);
-                gp->signal();
-                co_return;
-            });
-        BEAST_EXPECT(runner);
-        BEAST_EXPECT(g.wait_for(5s));
-        runner->join();
-        BEAST_EXPECT(result == 14);  // (3 + 4) * 2
-        BEAST_EXPECT(!runner->runnable());
-    }
-
-    /**
-     * postCoroTask returns nullptr when JobQueue is stopping.
-     */
-    void
-    testShutdownRejection()
-    {
-        using namespace std::chrono_literals;
-        using namespace jtx;
-
-        testcase("shutdown rejection");
-
-        Env env(*this, envconfig([](std::unique_ptr<Config> cfg) {
-            cfg->FORCE_MULTI_THREAD = true;
-            return cfg;
-        }));
-
-        // Stop the JobQueue
-        env.app().getJobQueue().stop();
-
-        auto runner = env.app().getJobQueue().postCoroTask(
-            jtCLIENT, "CoroTaskTest", [](auto) -> CoroTask<void> { co_return; });
-        BEAST_EXPECT(!runner);
-    }
-
-    void
-    run() override
-    {
-        testVoidCompletion();
-        testCorrectOrder();
-        testIncorrectOrder();
-        testJobQueueAwaiter();
-        testThreadSpecificStorage();
-        testExceptionPropagation();
-        testMultipleYields();
-        testValueReturn();
-        testValueException();
-        testValueChaining();
-        testShutdownRejection();
-    }
-};
-
-BEAST_DEFINE_TESTSUITE(CoroTask, core, xrpl);
-
-}  // namespace test
-}  // namespace xrpl

src/test/core/Coroutine_test.cpp

@@ -40,11 +40,6 @@ public:
         }
     };
 
-    // NOTE: All coroutine lambdas passed to postCoroTask use explicit
-    // pointer-by-value captures instead of [&] to work around a GCC 14
-    // bug where reference captures in coroutine lambdas are corrupted
-    // in the coroutine frame.
-
     void
     correct_order()
     {
@@ -59,15 +54,13 @@ public:
         }));
 
         gate g1, g2;
-        std::shared_ptr<JobQueue::CoroTaskRunner> c;
-        env.app().getJobQueue().postCoroTask(
-            jtCLIENT, "CoroTest", [cp = &c, g1p = &g1, g2p = &g2](auto runner) -> CoroTask<void> {
-                *cp = runner;
-                g1p->signal();
-                co_await runner->suspend();
-                g2p->signal();
-                co_return;
-            });
+        std::shared_ptr<JobQueue::Coro> c;
+        env.app().getJobQueue().postCoro(jtCLIENT, "CoroTest", [&](auto const& cr) {
+            c = cr;
+            g1.signal();
+            c->yield();
+            g2.signal();
+        });
         BEAST_EXPECT(g1.wait_for(5s));
         c->join();
         c->post();
@@ -88,17 +81,11 @@ public:
         }));
 
         gate g;
-        env.app().getJobQueue().postCoroTask(
-            jtCLIENT, "CoroTest", [gp = &g](auto runner) -> CoroTask<void> {
-                // Schedule a resume before suspending.  The posted job
-                // cannot actually call resume() until the current resume()
-                // releases CoroTaskRunner::mutex_, which only happens after
-                // the coroutine suspends at co_await.
-                runner->post();
-                co_await runner->suspend();
-                gp->signal();
-                co_return;
-            });
+        env.app().getJobQueue().postCoro(jtCLIENT, "CoroTest", [&](auto const& c) {
+            c->post();
+            c->yield();
+            g.signal();
+        });
         BEAST_EXPECT(g.wait_for(5s));
     }
 
@@ -114,7 +101,7 @@ public:
         auto& jq = env.app().getJobQueue();
 
         static int const N = 4;
-        std::array<std::shared_ptr<JobQueue::CoroTaskRunner>, N> a;
+        std::array<std::shared_ptr<JobQueue::Coro>, N> a;
 
         LocalValue<int> lv(-1);
         BEAST_EXPECT(*lv == -1);
@@ -131,23 +118,19 @@ public:
 
         for (int i = 0; i < N; ++i)
         {
-            jq.postCoroTask(
-                jtCLIENT,
-                "CoroTest",
-                [this, ap = &a, gp = &g, lvp = &lv, id = i](auto runner) -> CoroTask<void> {
-                    (*ap)[id] = runner;
-                    gp->signal();
-                    co_await runner->suspend();
+            jq.postCoro(jtCLIENT, "CoroTest", [&, id = i](auto const& c) {
+                a[id] = c;
+                g.signal();
+                c->yield();
 
-                    this->BEAST_EXPECT(**lvp == -1);
-                    **lvp = id;
-                    this->BEAST_EXPECT(**lvp == id);
-                    gp->signal();
-                    co_await runner->suspend();
+                this->BEAST_EXPECT(*lv == -1);
+                *lv = id;
+                this->BEAST_EXPECT(*lv == id);
+                g.signal();
+                c->yield();
 
-                    this->BEAST_EXPECT(**lvp == id);
-                    co_return;
-                });
+                this->BEAST_EXPECT(*lv == id);
+            });
             BEAST_EXPECT(g.wait_for(5s));
             a[i]->join();
         }

src/test/core/JobQueue_test.cpp

@@ -43,91 +43,87 @@ class JobQueue_test : public beast::unit_test::suite
         }
     }
 
-    // NOTE: All coroutine lambdas passed to postCoroTask use explicit
-    // pointer-by-value captures instead of [&] to work around a GCC 14
-    // bug where reference captures in coroutine lambdas are corrupted
-    // in the coroutine frame.
-
     void
-    testPostCoroTask()
+    testPostCoro()
     {
         jtx::Env env{*this};
 
         JobQueue& jQueue = env.app().getJobQueue();
         {
-            // Test repeated post()s until the coroutine completes.
+            // Test repeated post()s until the Coro completes.
             std::atomic<int> yieldCount{0};
-            auto const runner = jQueue.postCoroTask(
-                jtCLIENT, "PostCoroTest1", [ycp = &yieldCount](auto runner) -> CoroTask<void> {
-                    while (++(*ycp) < 4)
-                        co_await runner->suspend();
-                    co_return;
+            auto const coro = jQueue.postCoro(
+                jtCLIENT,
+                "PostCoroTest1",
+                [&yieldCount](std::shared_ptr<JobQueue::Coro> const& coroCopy) {
+                    while (++yieldCount < 4)
+                        coroCopy->yield();
                 });
-            BEAST_EXPECT(runner != nullptr);
+            BEAST_EXPECT(coro != nullptr);
 
             // Wait for the Job to run and yield.
             while (yieldCount == 0)
                 ;
 
-            // Now re-post until the CoroTaskRunner says it is done.
+            // Now re-post until the Coro says it is done.
             int old = yieldCount;
-            while (runner->runnable())
+            while (coro->runnable())
             {
-                BEAST_EXPECT(runner->post());
+                BEAST_EXPECT(coro->post());
                 while (old == yieldCount)
                 {
                 }
-                runner->join();
+                coro->join();
                 BEAST_EXPECT(++old == yieldCount);
             }
             BEAST_EXPECT(yieldCount == 4);
         }
         {
-            // Test repeated resume()s until the coroutine completes.
+            // Test repeated resume()s until the Coro completes.
             int yieldCount{0};
-            auto const runner = jQueue.postCoroTask(
-                jtCLIENT, "PostCoroTest2", [ycp = &yieldCount](auto runner) -> CoroTask<void> {
-                    while (++(*ycp) < 4)
-                        co_await runner->suspend();
-                    co_return;
+            auto const coro = jQueue.postCoro(
+                jtCLIENT,
+                "PostCoroTest2",
+                [&yieldCount](std::shared_ptr<JobQueue::Coro> const& coroCopy) {
+                    while (++yieldCount < 4)
+                        coroCopy->yield();
                 });
-            if (!runner)
+            if (!coro)
             {
-                // There's no good reason we should not get a runner, but we
+                // There's no good reason we should not get a Coro, but we
                 // can't continue without one.
                 BEAST_EXPECT(false);
                 return;
             }
 
             // Wait for the Job to run and yield.
-            runner->join();
+            coro->join();
 
-            // Now resume until the CoroTaskRunner says it is done.
+            // Now resume until the Coro says it is done.
             int old = yieldCount;
-            while (runner->runnable())
+            while (coro->runnable())
             {
-                runner->resume();  // Resume runs synchronously on this thread.
+                coro->resume();  // Resume runs synchronously on this thread.
                 BEAST_EXPECT(++old == yieldCount);
             }
             BEAST_EXPECT(yieldCount == 4);
         }
         {
             // If the JobQueue is stopped, we should no
-            // longer be able to post a coroutine (and calling postCoroTask()
-            // should return nullptr).
+            // longer be able to add a Coro (and calling postCoro() should
+            // return false).
             using namespace std::chrono_literals;
             jQueue.stop();
 
-            // The coroutine should never run, so having it access this
+            // The Coro should never run, so having the Coro access this
             // unprotected variable on the stack should be completely safe.
             // Not recommended for the faint of heart...
             bool unprotected;
-            auto const runner = jQueue.postCoroTask(
-                jtCLIENT, "PostCoroTest3", [up = &unprotected](auto) -> CoroTask<void> {
-                    *up = false;
-                    co_return;
+            auto const coro = jQueue.postCoro(
+                jtCLIENT, "PostCoroTest3", [&unprotected](std::shared_ptr<JobQueue::Coro> const&) {
+                    unprotected = false;
                 });
-            BEAST_EXPECT(runner == nullptr);
+            BEAST_EXPECT(coro == nullptr);
         }
     }
 
@@ -136,7 +132,7 @@ public:
     run() override
     {
         testAddJob();
-        testPostCoroTask();
+        testPostCoro();
     }
 };
 

src/test/jtx/envconfig.h

@@ -57,6 +57,17 @@ envconfig(F&& modfunc, Args&&... args)
     return modfunc(envconfig(), std::forward<Args>(args)...);
 }
 
+/// @brief adjust config to enable online_delete
+///
+/// @param cfg config instance to be modified
+///
+/// @param deleteInterval how many new ledgers should be available before
+/// rotating. Defaults to 8, because the standalone minimum is 8.
+///
+/// @return unique_ptr to Config instance
+std::unique_ptr<Config>
+online_delete(std::unique_ptr<Config> cfg, std::uint32_t deleteInterval = 8);
+
 /// @brief adjust config so no admin ports are enabled
 ///
 /// this is intended for use with envconfig, as in

src/test/jtx/impl/AMMTest.cpp

@@ -6,7 +6,6 @@
 
 #include <xrpld/rpc/RPCHandler.h>
 
-#include <xrpl/core/CoroTask.h>
 #include <xrpl/protocol/ApiVersion.h>
 #include <xrpl/protocol/STParsedJSON.h>
 #include <xrpl/resource/Fees.h>
@@ -194,6 +193,7 @@ AMMTest::find_paths_request(
          c,
          Role::USER,
          {},
+         {},
          RPC::apiVersionIfUnspecified},
         {},
         {}};
@@ -215,11 +215,11 @@ AMMTest::find_paths_request(
 
     Json::Value result;
     gate g;
-    app.getJobQueue().postCoroTask(jtCLIENT, "RPC-Client", [&](auto) -> CoroTask<void> {
+    app.getJobQueue().postCoro(jtCLIENT, "RPC-Client", [&](auto const& coro) {
         context.params = std::move(params);
+        context.coro = coro;
         RPC::doCommand(context, result);
         g.signal();
-        co_return;
     });
 
     using namespace std::chrono_literals;

src/test/jtx/impl/envconfig.cpp

@@ -53,6 +53,15 @@ setupConfigForUnitTests(Config& cfg)
 
 namespace jtx {
 
+std::unique_ptr<Config>
+online_delete(std::unique_ptr<Config> cfg, std::uint32_t deleteInterval)
+{
+    cfg->LEDGER_HISTORY = deleteInterval;
+    auto& section = cfg->section(ConfigSection::nodeDatabase());
+    section.set("online_delete", std::to_string(deleteInterval));
+    return cfg;
+}
+
 std::unique_ptr<Config>
 no_admin(std::unique_ptr<Config> cfg)
 {

src/test/nodestore/TestBase.h

@@ -138,7 +138,7 @@ public:
         {
             std::shared_ptr<NodeObject> object;
 
-            Status const status = backend.fetch(batch[i]->getHash(), &object);
+            Status const status = backend.fetch(batch[i]->getHash().cbegin(), &object);
 
             BEAST_EXPECT(status == ok);
 
@@ -158,7 +158,7 @@ public:
         {
             std::shared_ptr<NodeObject> object;
 
-            Status const status = backend.fetch(batch[i]->getHash(), &object);
+            Status const status = backend.fetch(batch[i]->getHash().cbegin(), &object);
 
             BEAST_EXPECT(status == notFound);
         }

src/test/nodestore/Timing_test.cpp

@@ -314,7 +314,7 @@ public:
                     std::shared_ptr<NodeObject> obj;
                     std::shared_ptr<NodeObject> result;
                     obj = seq1_.obj(dist_(gen_));
-                    backend_.fetch(obj->getHash(), &result);
+                    backend_.fetch(obj->getHash().data(), &result);
                     suite_.expect(result && isSame(result, obj));
                 }
                 catch (std::exception const& e)
@@ -377,9 +377,9 @@ public:
             {
                 try
                 {
-                    auto const hash = seq2_.key(i);
+                    auto const key = seq2_.key(i);
                     std::shared_ptr<NodeObject> result;
-                    backend_.fetch(hash, &result);
+                    backend_.fetch(key.data(), &result);
                     suite_.expect(!result);
                 }
                 catch (std::exception const& e)
@@ -449,9 +449,9 @@ public:
                 {
                     if (rand_(gen_) < missingNodePercent)
                     {
-                        auto const hash = seq2_.key(dist_(gen_));
+                        auto const key = seq2_.key(dist_(gen_));
                         std::shared_ptr<NodeObject> result;
-                        backend_.fetch(hash, &result);
+                        backend_.fetch(key.data(), &result);
                         suite_.expect(!result);
                     }
                     else
@@ -459,7 +459,7 @@ public:
                         std::shared_ptr<NodeObject> obj;
                         std::shared_ptr<NodeObject> result;
                         obj = seq1_.obj(dist_(gen_));
-                        backend_.fetch(obj->getHash(), &result);
+                        backend_.fetch(obj->getHash().data(), &result);
                         suite_.expect(result && isSame(result, obj));
                     }
                 }
@@ -540,7 +540,8 @@ public:
                         std::shared_ptr<NodeObject> result;
                         auto const j = older_(gen_);
                         obj = seq1_.obj(j);
-                        backend_.fetch(obj->getHash(), &result);
+                        std::shared_ptr<NodeObject> result1;
+                        backend_.fetch(obj->getHash().data(), &result);
                         suite_.expect(result != nullptr);
                         suite_.expect(isSame(result, obj));
                     }
@@ -558,7 +559,7 @@ public:
                                 std::shared_ptr<NodeObject> result;
                                 auto const j = recent_(gen_);
                                 obj = seq1_.obj(j);
-                                backend_.fetch(obj->getHash(), &result);
+                                backend_.fetch(obj->getHash().data(), &result);
                                 suite_.expect(!result || isSame(result, obj));
                                 break;
                             }

src/xrpld/app/ledger/LedgerMaster.h

@@ -104,7 +104,10 @@ public:
     failedSave(std::uint32_t seq, uint256 const& hash);
 
     std::string
-    getCompleteLedgers();
+    getCompleteLedgers() const;
+
+    std::size_t
+    missingFromCompleteLedgerRange(LedgerIndex first, LedgerIndex last) const;
 
     /** Apply held transactions to the open ledger
         This is normally called as we close the ledger.
@@ -321,7 +324,7 @@ private:
     // A set of transactions to replay during the next close
     std::unique_ptr<LedgerReplay> replayData;
 
-    std::recursive_mutex mCompleteLock;
+    std::recursive_mutex mutable mCompleteLock;
     RangeSet<std::uint32_t> mCompleteLedgers;
 
     // Publish thread is running.

src/xrpld/app/ledger/detail/LedgerMaster.cpp

@@ -1510,12 +1510,34 @@ LedgerMaster::getPublishedLedger()
 }
 
 std::string
-LedgerMaster::getCompleteLedgers()
+LedgerMaster::getCompleteLedgers() const
 {
     std::lock_guard sl(mCompleteLock);
     return to_string(mCompleteLedgers);
 }
 
+std::size_t
+LedgerMaster::missingFromCompleteLedgerRange(LedgerIndex first, LedgerIndex last) const
+{
+    // Make a copy of the range to avoid holding the lock
+    auto const range = [&] {
+        std::lock_guard sl(mCompleteLock);
+        return mCompleteLedgers;
+    }();
+
+    std::size_t missing = 0;
+
+    for (LedgerIndex idx = first; idx <= last; ++idx)
+    {
+        if (!boost::icl::contains(range, idx))
+        {
+            ++missing;
+        }
+    }
+
+    return missing;
+}
+
 std::optional<NetClock::time_point>
 LedgerMaster::getCloseTimeBySeq(LedgerIndex ledgerIndex)
 {

src/xrpld/app/main/Application.cpp

@@ -1425,6 +1425,7 @@ ApplicationImp::setup(boost::program_options::variables_map const& cmdline)
              c,
              Role::ADMIN,
              {},
+             {},
              RPC::apiMaximumSupportedVersion},
             jvCommand};
 

src/xrpld/app/main/GRPCServer.cpp

@@ -3,7 +3,6 @@
 
 #include <xrpl/beast/core/CurrentThreadName.h>
 #include <xrpl/beast/net/IPAddressConversion.h>
-#include <xrpl/core/CoroTask.h>
 #include <xrpl/resource/Fees.h>
 
 namespace xrpl {
@@ -100,14 +99,13 @@ GRPCServerImpl::CallData<Request, Response>::process()
     // ensures that finished is always true when this CallData object
     // is returned as a tag in handleRpcs(), after sending the response
     finished_ = true;
-    auto runner = app_.getJobQueue().postCoroTask(
-        JobType::jtRPC, "gRPC-Client", [thisShared](auto) -> CoroTask<void> {
-            thisShared->processRequest();
-            co_return;
+    auto coro = app_.getJobQueue().postCoro(
+        JobType::jtRPC, "gRPC-Client", [thisShared](std::shared_ptr<JobQueue::Coro> coro) {
+            thisShared->process(coro);
         });
 
-    // If runner is null, then the JobQueue has already been shutdown
-    if (!runner)
+    // If coro is null, then the JobQueue has already been shutdown
+    if (!coro)
     {
         grpc::Status status{grpc::StatusCode::INTERNAL, "Job Queue is already stopped"};
         responder_.FinishWithError(status, this);
@@ -116,7 +114,7 @@ GRPCServerImpl::CallData<Request, Response>::process()
 
 template <class Request, class Response>
 void
-GRPCServerImpl::CallData<Request, Response>::processRequest()
+GRPCServerImpl::CallData<Request, Response>::process(std::shared_ptr<JobQueue::Coro> coro)
 {
     try
     {
@@ -158,6 +156,7 @@ GRPCServerImpl::CallData<Request, Response>::processRequest()
                  app_.getLedgerMaster(),
                  usage,
                  role,
+                 coro,
                  InfoSub::pointer(),
                  apiVersion},
                 request_};

src/xrpld/app/main/GRPCServer.h

@@ -206,12 +206,9 @@ private:
         clone() override;
 
     private:
-        /**
-         * Process the gRPC request. Called inside the CoroTask lambda
-         * posted to the JobQueue by process().
-         */
+        // process the request. Called inside the coroutine passed to JobQueue
         void
-        processRequest();
+        process(std::shared_ptr<JobQueue::Coro> coro);
 
         // return load type of this RPC
         Resource::Charge

src/xrpld/app/misc/SHAMapStoreImp.cpp

@@ -261,6 +261,17 @@ SHAMapStoreImp::run()
         bool const readyToRotate = validatedSeq >= lastRotated + deleteInterval_ &&
             canDelete_ >= lastRotated - 1 && healthWait() == keepGoing;
 
+        JLOG(journal_.debug()) << "run: Setting lastGoodValidatedLedger_ to " << validatedSeq;
+
+        {
+            // Note that this is set after the healthWait() check, so that we
+            // don't start the rotation until the validated ledger is fully
+            // processed. It is not guaranteed to be done at this point. It also
+            // allows the testLedgerGaps unit test to work.
+            std::unique_lock<std::mutex> lock(mutex_);
+            lastGoodValidatedLedger_ = validatedSeq;
+        }
+
         // will delete up to (not including) lastRotated
         if (readyToRotate)
         {
@@ -268,7 +279,8 @@ SHAMapStoreImp::run()
                                   << lastRotated << " deleteInterval " << deleteInterval_
                                   << " canDelete_ " << canDelete_ << " state "
                                   << app_.getOPs().strOperatingMode(false) << " age "
-                                  << ledgerMaster_->getValidatedLedgerAge().count() << 's';
+                                  << ledgerMaster_->getValidatedLedgerAge().count()
+                                  << "s. Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
 
             clearPrior(lastRotated);
             if (healthWait() == stopping)
@@ -328,7 +340,9 @@ SHAMapStoreImp::run()
                     clearCaches(validatedSeq);
                 });
 
-            JLOG(journal_.warn()) << "finished rotation " << validatedSeq;
+            JLOG(journal_.warn()) << "finished rotation. validatedSeq: " << validatedSeq
+                                  << ", lastRotated: " << lastRotated
+                                  << ". Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
         }
     }
 }
@@ -562,22 +576,40 @@ SHAMapStoreImp::clearPrior(LedgerIndex lastRotated)
 SHAMapStoreImp::HealthResult
 SHAMapStoreImp::healthWait()
 {
+    auto index = ledgerMaster_->getValidLedgerIndex();
     auto age = ledgerMaster_->getValidatedLedgerAge();
     OperatingMode mode = netOPs_->getOperatingMode();
     std::unique_lock lock(mutex_);
-    while (!stop_ && (mode != OperatingMode::FULL || age > ageThreshold_))
+
+    auto numMissing =
+        ledgerMaster_->missingFromCompleteLedgerRange(lastGoodValidatedLedger_, index);
+    while (!stop_ && (mode != OperatingMode::FULL || age > ageThreshold_ || numMissing > 0))
     {
+        // this value shouldn't change, so grab it while we have the
+        // lock
+        auto const lastGood = lastGoodValidatedLedger_;
+
         lock.unlock();
-        JLOG(journal_.warn()) << "Waiting " << recoveryWaitTime_.count()
-                              << "s for node to stabilize. state: "
-                              << app_.getOPs().strOperatingMode(mode, false) << ". age "
-                              << age.count() << 's';
+
+        auto const stream =
+            mode != OperatingMode::FULL || age > ageThreshold_ ? journal_.warn() : journal_.info();
+        JLOG(stream) << "Waiting " << recoveryWaitTime_.count()
+                     << "s for node to stabilize. state: "
+                     << app_.getOPs().strOperatingMode(mode, false) << ". age " << age.count()
+                     << "s. Missing ledgers: " << numMissing << ". Expect: " << lastGood << "-"
+                     << index << ". Complete ledgers: " << ledgerMaster_->getCompleteLedgers();
         std::this_thread::sleep_for(recoveryWaitTime_);
+        index = ledgerMaster_->getValidLedgerIndex();
         age = ledgerMaster_->getValidatedLedgerAge();
         mode = netOPs_->getOperatingMode();
+        numMissing = ledgerMaster_->missingFromCompleteLedgerRange(lastGood, index);
+
         lock.lock();
     }
 
+    JLOG(journal_.debug()) << "healthWait: Setting lastGoodValidatedLedger_ to " << index;
+    lastGoodValidatedLedger_ = index;
+
     return stop_ ? stopping : keepGoing;
 }
 

src/xrpld/app/misc/SHAMapStoreImp.h

@@ -70,6 +70,11 @@ private:
     std::thread thread_;
     bool stop_ = false;
     bool healthy_ = true;
+    // Used to prevent ledger gaps from forming during online deletion. Keeps
+    // track of the last validated ledger that was processed without gaps. There
+    // are no guarantees about gaps while online delete is not running. For
+    // that, use advisory_delete and check for gaps externally.
+    LedgerIndex lastGoodValidatedLedger_ = 0;
     mutable std::condition_variable cond_;
     mutable std::condition_variable rendezvous_;
     mutable std::mutex mutex_;
@@ -83,11 +88,11 @@ private:
     std::uint32_t deleteBatch_ = 100;
     std::chrono::milliseconds backOff_{100};
     std::chrono::seconds ageThreshold_{60};
-    /// If  the node is out of sync during an online_delete healthWait()
-    /// call, sleep the thread for this time, and continue checking until
-    /// recovery.
+    /// If  the node is out of sync, or any recent ledgers are not
+    /// available during an online_delete healthWait() call, sleep
+    /// the thread for this time, and continue checking until recovery.
     /// See also: "recovery_wait_seconds" in xrpld-example.cfg
-    std::chrono::seconds recoveryWaitTime_{5};
+    std::chrono::seconds recoveryWaitTime_{1};
 
     // these do not exist upon SHAMapStore creation, but do exist
     // as of run() or before
@@ -204,6 +209,8 @@ private:
     enum HealthResult { stopping, keepGoing };
     [[nodiscard]] HealthResult
     healthWait();
+    bool
+    hasCompleteRange(LedgerIndex first, LedgerIndex last);
 
 public:
     void

src/xrpld/rpc/Context.h

@@ -3,6 +3,7 @@
 #include <xrpld/rpc/Role.h>
 
 #include <xrpl/beast/utility/Journal.h>
+#include <xrpl/core/JobQueue.h>
 #include <xrpl/server/InfoSub.h>
 
 namespace xrpl {
@@ -23,6 +24,7 @@ struct Context
     LedgerMaster& ledgerMaster;
     Resource::Consumer& consumer;
     Role role;
+    std::shared_ptr<JobQueue::Coro> coro{};
     InfoSub::pointer infoSub{};
     unsigned int apiVersion;
 };

src/xrpld/rpc/ServerHandler.h

@@ -169,10 +169,13 @@ public:
 
 private:
     Json::Value
-    processSession(std::shared_ptr<WSSession> const& session, Json::Value const& jv);
+    processSession(
+        std::shared_ptr<WSSession> const& session,
+        std::shared_ptr<JobQueue::Coro> const& coro,
+        Json::Value const& jv);
 
     void
-    processSession(std::shared_ptr<Session> const&);
+    processSession(std::shared_ptr<Session> const&, std::shared_ptr<JobQueue::Coro> coro);
 
     void
     processRequest(
@@ -180,6 +183,7 @@ private:
         std::string const& request,
         beast::IP::Endpoint const& remoteIPAddress,
         Output&&,
+        std::shared_ptr<JobQueue::Coro> coro,
         std::string_view forwardedFor,
         std::string_view user);
 

src/xrpld/rpc/detail/ServerHandler.cpp

@@ -14,7 +14,6 @@
 #include <xrpl/basics/make_SSLContext.h>
 #include <xrpl/beast/net/IPAddressConversion.h>
 #include <xrpl/beast/rfc2616.h>
-#include <xrpl/core/CoroTask.h>
 #include <xrpl/core/JobQueue.h>
 #include <xrpl/json/json_reader.h>
 #include <xrpl/json/to_string.h>
@@ -285,10 +284,9 @@ ServerHandler::onRequest(Session& session)
     }
 
     std::shared_ptr<Session> detachedSession = session.detach();
-    auto const postResult = m_jobQueue.postCoroTask(
-        jtCLIENT_RPC, "RPC-Client", [this, detachedSession](auto) -> CoroTask<void> {
-            processSession(detachedSession);
-            co_return;
+    auto const postResult = m_jobQueue.postCoro(
+        jtCLIENT_RPC, "RPC-Client", [this, detachedSession](std::shared_ptr<JobQueue::Coro> coro) {
+            processSession(detachedSession, coro);
         });
     if (postResult == nullptr)
     {
@@ -324,18 +322,17 @@ ServerHandler::onWSMessage(
 
     JLOG(m_journal.trace()) << "Websocket received '" << jv << "'";
 
-    auto const postResult = m_jobQueue.postCoroTask(
+    auto const postResult = m_jobQueue.postCoro(
         jtCLIENT_WEBSOCKET,
         "WS-Client",
-        [this, session, jv = std::move(jv)](auto) -> CoroTask<void> {
-            auto const jr = this->processSession(session, jv);
+        [this, session, jv = std::move(jv)](std::shared_ptr<JobQueue::Coro> const& coro) {
+            auto const jr = this->processSession(session, coro, jv);
             auto const s = to_string(jr);
             auto const n = s.length();
             boost::beast::multi_buffer sb(n);
             sb.commit(boost::asio::buffer_copy(sb.prepare(n), boost::asio::buffer(s.c_str(), n)));
             session->send(std::make_shared<StreambufWSMsg<decltype(sb)>>(std::move(sb)));
             session->complete();
-            co_return;
         });
     if (postResult == nullptr)
     {
@@ -376,7 +373,10 @@ logDuration(Json::Value const& request, T const& duration, beast::Journal& journ
 }
 
 Json::Value
-ServerHandler::processSession(std::shared_ptr<WSSession> const& session, Json::Value const& jv)
+ServerHandler::processSession(
+    std::shared_ptr<WSSession> const& session,
+    std::shared_ptr<JobQueue::Coro> const& coro,
+    Json::Value const& jv)
 {
     auto is = std::static_pointer_cast<WSInfoSub>(session->appDefined);
     if (is->getConsumer().disconnect(m_journal))
@@ -443,6 +443,7 @@ ServerHandler::processSession(std::shared_ptr<WSSession> const& session, Json::V
                  app_.getLedgerMaster(),
                  is->getConsumer(),
                  role,
+                 coro,
                  is,
                  apiVersion},
                 jv,
@@ -513,14 +514,18 @@ ServerHandler::processSession(std::shared_ptr<WSSession> const& session, Json::V
     return jr;
 }
 
+// Run as a coroutine.
 void
-ServerHandler::processSession(std::shared_ptr<Session> const& session)
+ServerHandler::processSession(
+    std::shared_ptr<Session> const& session,
+    std::shared_ptr<JobQueue::Coro> coro)
 {
     processRequest(
         session->port(),
         buffers_to_string(session->request().body().data()),
         session->remoteAddress().at_port(0),
         makeOutput(*session),
+        coro,
         forwardedFor(session->request()),
         [&] {
             auto const iter = session->request().find("X-User");
@@ -557,6 +562,7 @@ ServerHandler::processRequest(
     std::string const& request,
     beast::IP::Endpoint const& remoteIPAddress,
     Output&& output,
+    std::shared_ptr<JobQueue::Coro> coro,
     std::string_view forwardedFor,
     std::string_view user)
 {
@@ -813,6 +819,7 @@ ServerHandler::processRequest(
              app_.getLedgerMaster(),
              usage,
              role,
+             coro,
              InfoSub::pointer(),
              apiVersion},
             params,

src/xrpld/rpc/handlers/RipplePathFind.cpp

@@ -7,9 +7,6 @@
 #include <xrpl/protocol/RPCErr.h>
 #include <xrpl/resource/Fees.h>
 
-#include <condition_variable>
-#include <mutex>
-
 namespace xrpl {
 
 // This interface is deprecated.
@@ -40,31 +37,98 @@ doRipplePathFind(RPC::JsonContext& context)
         PathRequest::pointer request;
         lpLedger = context.ledgerMaster.getClosedLedger();
 
-        // makeLegacyPathRequest enqueues a path-finding job that runs
-        // asynchronously.  We block this thread with a condition_variable
-        // until the path-finding continuation signals completion.
-        // If makeLegacyPathRequest cannot schedule the job (e.g. during
-        // shutdown), it returns an empty request and we skip the wait.
-        std::mutex mtx;
-        std::condition_variable cv;
-        bool pathDone = false;
-
+        // It doesn't look like there's much odd happening here, but you should
+        // be aware this code runs in a JobQueue::Coro, which is a coroutine.
+        // And we may be flipping around between threads.  Here's an overview:
+        //
+        // 1. We're running doRipplePathFind() due to a call to
+        //    ripple_path_find.  doRipplePathFind() is currently running
+        //    inside of a JobQueue::Coro using a JobQueue thread.
+        //
+        // 2. doRipplePathFind's call to makeLegacyPathRequest() enqueues the
+        //    path-finding request.  That request will (probably) run at some
+        //    indeterminate future time on a (probably different) JobQueue
+        //    thread.
+        //
+        // 3. As a continuation from that path-finding JobQueue thread, the
+        //    coroutine we're currently running in (!) is posted to the
+        //    JobQueue.  Because it is a continuation, that post won't
+        //    happen until the path-finding request completes.
+        //
+        // 4. Once the continuation is enqueued, and we have reason to think
+        //    the path-finding job is likely to run, then the coroutine we're
+        //    running in yield()s.  That means it surrenders its thread in
+        //    the JobQueue.  The coroutine is suspended, but ready to run,
+        //    because it is kept resident by a shared_ptr in the
+        //    path-finding continuation.
+        //
+        // 5. If all goes well then path-finding runs on a JobQueue thread
+        //    and executes its continuation.  The continuation posts this
+        //    same coroutine (!) to the JobQueue.
+        //
+        // 6. When the JobQueue calls this coroutine, this coroutine resumes
+        //    from the line below the coro->yield() and returns the
+        //    path-finding result.
+        //
+        // With so many moving parts, what could go wrong?
+        //
+        // Just in terms of the JobQueue refusing to add jobs at shutdown
+        // there are two specific things that can go wrong.
+        //
+        // 1. The path-finding Job queued by makeLegacyPathRequest() might be
+        //    rejected (because we're shutting down).
+        //
+        //    Fortunately this problem can be addressed by looking at the
+        //    return value of makeLegacyPathRequest().  If
+        //    makeLegacyPathRequest() cannot get a thread to run the path-find
+        //    on, then it returns an empty request.
+        //
+        // 2. The path-finding job might run, but the Coro::post() might be
+        //    rejected by the JobQueue (because we're shutting down).
+        //
+        //    We handle this case by resuming (not posting) the Coro.
+        //    By resuming the Coro, we allow the Coro to run to completion
+        //    on the current thread instead of requiring that it run on a
+        //    new thread from the JobQueue.
+        //
+        // Both of these failure modes are hard to recreate in a unit test
+        // because they are so dependent on inter-thread timing.  However
+        // the failure modes can be observed by synchronously (inside the
+        // rippled source code) shutting down the application.  The code to
+        // do so looks like this:
+        //
+        //   context.app.signalStop();
+        //   while (! context.app.getJobQueue().jobCounter().joined()) { }
+        //
+        // The first line starts the process of shutting down the app.
+        // The second line waits until no more jobs can be added to the
+        // JobQueue before letting the thread continue.
+        //
+        // May 2017
         jvResult = context.app.getPathRequests().makeLegacyPathRequest(
             request,
-            [&]() {
+            [&context]() {
+                // Copying the shared_ptr keeps the coroutine alive up
+                // through the return.  Otherwise the storage under the
+                // captured reference could evaporate when we return from
+                // coroCopy->resume().  This is not strictly necessary, but
+                // will make maintenance easier.
+                std::shared_ptr<JobQueue::Coro> coroCopy{context.coro};
+                if (!coroCopy->post())
                 {
-                    std::lock_guard lk(mtx);
-                    pathDone = true;
+                    // The post() failed, so we won't get a thread to let
+                    // the Coro finish.  We'll call Coro::resume() so the
+                    // Coro can finish on our thread.  Otherwise the
+                    // application will hang on shutdown.
+                    coroCopy->resume();
                 }
-                cv.notify_one();
             },
             context.consumer,
             lpLedger,
             context.params);
         if (request)
         {
-            std::unique_lock lk(mtx);
-            cv.wait(lk, [&] { return pathDone; });
+            context.coro->yield();
             jvResult = request->doStatus(context.params);
         }