feat: Add parallel apply access-set scheduler

Co-authored-by: Bart <11445373+bthomee@users.noreply.github.com>

.github/actions/build-deps/action.yml

@@ -37,12 +37,12 @@ runs:
       run: |
         echo 'Installing dependencies.'
         conan install \
-          --profile ci \
-          --build="${BUILD_OPTION}" \
-          --options:host='&:tests=True' \
-          --options:host='&:xrpld=True' \
-          --settings:all build_type="${BUILD_TYPE}" \
-          --conf:all tools.build:jobs=${BUILD_NPROC} \
-          --conf:all tools.build:verbosity="${LOG_VERBOSITY}" \
-          --conf:all tools.compilation:verbosity="${LOG_VERBOSITY}" \
-          .
+            --profile ci \
+            --build="${BUILD_OPTION}" \
+            --options:host='&:tests=True' \
+            --options:host='&:xrpld=True' \
+            --settings:all build_type="${BUILD_TYPE}" \
+            --conf:all tools.build:jobs=${BUILD_NPROC} \
+            --conf:all tools.build:verbosity="${LOG_VERBOSITY}" \
+            --conf:all tools.compilation:verbosity="${LOG_VERBOSITY}" \
+            .

.github/actions/generate-version/action.yml

@@ -15,7 +15,7 @@ runs:
       shell: bash
       env:
         VERSION: ${{ github.ref_name }}
-      run: echo "VERSION=${VERSION}" >> "${GITHUB_ENV}"
+      run: echo "VERSION=${VERSION}" >>"${GITHUB_ENV}"
 
     # When a tag is not pushed, then the version (e.g. 1.2.3-b0) is extracted
     # from the BuildInfo.cpp file and the shortened commit hash appended to it.
@@ -28,17 +28,17 @@ runs:
         echo 'Extracting version from BuildInfo.cpp.'
         VERSION="$(cat src/libxrpl/protocol/BuildInfo.cpp | grep "versionString =" | awk -F '"' '{print $2}')"
         if [[ -z "${VERSION}" ]]; then
-          echo 'Unable to extract version from BuildInfo.cpp.'
-          exit 1
+            echo 'Unable to extract version from BuildInfo.cpp.'
+            exit 1
         fi
 
         echo 'Appending shortened commit hash to version.'
         SHA='${{ github.sha }}'
         VERSION="${VERSION}+${SHA:0:7}"
 
-        echo "VERSION=${VERSION}" >> "${GITHUB_ENV}"
+        echo "VERSION=${VERSION}" >>"${GITHUB_ENV}"
 
     - name: Output version
       id: version
       shell: bash
-      run: echo "version=${VERSION}" >> "${GITHUB_OUTPUT}"
+      run: echo "version=${VERSION}" >>"${GITHUB_OUTPUT}"

.github/scripts/format-inline-bash.py

@@ -0,0 +1,403 @@
+#!/usr/bin/env python3
+
+"""
+Format embedded shell snippets using the shfmt hook configured in
+.pre-commit-config.yaml.
+
+Two shapes are recognised:
+
+* YAML workflow/action files: literal block-scalar runs (`run: |`) and
+  single-line runs (`run: some command`). A single-line run is upgraded to
+  a `run: |` block scalar if shfmt's output spans multiple lines.
+
+* Markdown files: ``` ```bash ``` fenced code blocks.
+
+Any block that shfmt cannot parse is skipped with a warning on stderr, so
+the file is left untouched and surrounding blocks still get formatted.
+
+For each occurrence the body is dedented, written to a temp .sh file,
+formatted via `pre-commit run shfmt --files <temp>` (falling back to
+`prek`), then re-indented and written back in place.
+
+When invoked without arguments, every .yml/.yaml under .github/ plus every
+.md file in the repo is scanned. When invoked with file arguments (the
+pre-commit case), only those files are processed.
+"""
+
+from __future__ import annotations
+
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Union
+
+REPO = Path(__file__).resolve().parents[2]
+
+_HOOK_RUNNER = next((cmd for cmd in ("pre-commit", "prek") if shutil.which(cmd)), None)
+if _HOOK_RUNNER is None:
+    sys.exit("error: neither `pre-commit` nor `prek` found on PATH")
+
+RUN_BLOCK_RE = re.compile(r"^(?P<prefix>[ \t]*(?:- )?)run:[ \t]*\|[+-]?[ \t]*$")
+RUN_INLINE_RE = re.compile(
+    r"^(?P<prefix>[ \t]*(?:- )?)run:[ \t]+" r"(?P<value>(?!\|[+-]?[ \t]*$)\S.*?)[ \t]*$"
+)
+MD_BASH_OPEN_RE = re.compile(r"^(?P<indent>[ ]{0,3})`{3}bash[ \t]*$")
+MD_FENCE_CLOSE_RE = re.compile(r"^[ ]{0,3}`{3,}[ \t]*$")
+
+
+@dataclass(frozen=True)
+class BlockRun:
+    """A `run: |` block scalar; `body_start:body_end` slices into `lines`."""
+
+    body_start: int
+    body_end: int
+    body_indent: int
+
+
+@dataclass(frozen=True)
+class InlineRun:
+    """A single-line `run: value` at `line_idx`."""
+
+    line_idx: int
+    prefix: str
+    value: str
+
+
+@dataclass(frozen=True)
+class MdBashBlock:
+    """A markdown ``` ```bash ``` fenced code block.
+
+    `body_start:body_end` slices into the file's lines; `open_line_idx`
+    points at the opening fence line.
+    """
+
+    open_line_idx: int
+    body_start: int
+    body_end: int
+    body_indent: int
+
+
+RunItem = Union[BlockRun, InlineRun]
+
+
+def _scan_block_body(
+    lines: list[str], body_start: int, run_col: int
+) -> tuple[int | None, int]:
+    """Locate the body of a `run: |` block scalar starting at `body_start`.
+
+    Returns `(body_indent, scan_end)`. `scan_end` is the line index where the
+    outer scanner should resume. `body_indent` is `None` when no body is
+    present (the scalar is empty, or the next non-blank line has indent
+    `<= run_col`).
+    """
+    body_indent: int | None = None
+    scan_end = len(lines)
+    for idx in range(body_start, len(lines)):
+        line = lines[idx]
+        if line.strip() == "":
+            continue
+        indent = len(line) - len(line.lstrip(" "))
+        if body_indent is None:
+            if indent > run_col:
+                body_indent = indent
+            else:
+                scan_end = idx
+                break
+        elif indent < body_indent:
+            scan_end = idx
+            break
+    if body_indent is not None:
+        while scan_end > body_start and lines[scan_end - 1].strip() == "":
+            scan_end -= 1
+        if scan_end <= body_start:
+            body_indent = None
+    return body_indent, scan_end
+
+
+def find_run_blocks(lines: list[str]) -> list[RunItem]:
+    """Return run items in document order."""
+    items: list[RunItem] = []
+    line_idx = 0
+    while line_idx < len(lines):
+        line = lines[line_idx]
+        if block_match := RUN_BLOCK_RE.match(line):
+            run_col = len(block_match.group("prefix"))
+            body_start = line_idx + 1
+            body_indent, scan_end = _scan_block_body(lines, body_start, run_col)
+            if body_indent is not None:
+                items.append(
+                    BlockRun(
+                        body_start=body_start,
+                        body_end=scan_end,
+                        body_indent=body_indent,
+                    )
+                )
+            line_idx = scan_end
+            continue
+        if inline_match := RUN_INLINE_RE.match(line):
+            items.append(
+                InlineRun(
+                    line_idx=line_idx,
+                    prefix=inline_match.group("prefix"),
+                    value=inline_match.group("value"),
+                )
+            )
+        line_idx += 1
+    return items
+
+
+def find_md_bash_blocks(lines: list[str]) -> list[MdBashBlock]:
+    """Return ``` ```bash ``` fenced code blocks in document order."""
+    blocks: list[MdBashBlock] = []
+    line_idx = 0
+    while line_idx < len(lines):
+        open_match = MD_BASH_OPEN_RE.match(lines[line_idx])
+        if not open_match:
+            line_idx += 1
+            continue
+        body_start = line_idx + 1
+        close_idx = next(
+            (
+                j
+                for j in range(body_start, len(lines))
+                if MD_FENCE_CLOSE_RE.match(lines[j])
+            ),
+            None,
+        )
+        if close_idx is None:
+            line_idx = body_start
+            continue
+        body = lines[body_start:close_idx]
+        non_blank = [b for b in body if b.strip()]
+        body_indent = (
+            min(len(b) - len(b.lstrip(" ")) for b in non_blank)
+            if non_blank
+            else len(open_match.group("indent"))
+        )
+        blocks.append(
+            MdBashBlock(
+                open_line_idx=line_idx,
+                body_start=body_start,
+                body_end=close_idx,
+                body_indent=body_indent,
+            )
+        )
+        line_idx = close_idx + 1
+    return blocks
+
+
+def dedent(lines: list[str], n: int) -> list[str]:
+    pad = " " * n
+    return [
+        (
+            ""
+            if line.strip() == ""
+            else (line[n:] if line.startswith(pad) else line.lstrip(" "))
+        )
+        for line in lines
+    ]
+
+
+def reindent(lines: list[str], n: int) -> list[str]:
+    pad = " " * n
+    return [pad + line if line else "" for line in lines]
+
+
+_SHFMT_ERR_RE = re.compile(r"\.sh:\d+:\d+:\s")
+_GHA_EXPR_RE = re.compile(r"\$\{\{.*?\}\}", re.DOTALL)
+_GHA_PLACEHOLDER_RE = re.compile(r"__GHA_EXPR_(\d+)__")
+
+
+def _encode_gha_exprs(text: str) -> tuple[str, list[str]]:
+    """Replace `${{ ... }}` expressions with bash-safe placeholder identifiers."""
+    exprs: list[str] = []
+
+    def repl(match: re.Match[str]) -> str:
+        exprs.append(match.group(0))
+        return f"__GHA_EXPR_{len(exprs) - 1}__"
+
+    return _GHA_EXPR_RE.sub(repl, text), exprs
+
+
+def _decode_gha_exprs(text: str, exprs: list[str]) -> str:
+    """Restore `${{ ... }}` expressions from placeholder identifiers."""
+    return _GHA_PLACEHOLDER_RE.sub(lambda m: exprs[int(m.group(1))], text)
+
+
+def shfmt_via_hook(tmp_path: Path) -> tuple[bool, str]:
+    # `${{ ... }}` is not valid shell, so swap it for a placeholder identifier
+    # that shfmt can parse, then restore it after formatting.
+    encoded, exprs = _encode_gha_exprs(tmp_path.read_text())
+    if exprs:
+        tmp_path.write_text(encoded)
+    res = subprocess.run(
+        [_HOOK_RUNNER, "run", "shfmt", "--files", str(tmp_path)],
+        cwd=REPO,
+        capture_output=True,
+        text=True,
+    )
+    output = res.stdout + res.stderr
+    # shfmt emits parse errors as "<path>:<line>:<col>: <message>".
+    parse_err = bool(_SHFMT_ERR_RE.search(output))
+    # A non-zero exit that is neither a parse error nor pre-commit's "I had
+    # to modify files" signal means the hook itself failed to run (missing
+    # binary, install failure, bad config, ...). Surface that loudly rather
+    # than silently treating it as a no-op.
+    if (
+        res.returncode != 0
+        and not parse_err
+        and "files were modified by this hook" not in output
+    ):
+        sys.exit(
+            f"error: `{_HOOK_RUNNER} run shfmt` failed with exit {res.returncode}:\n{output}"
+        )
+    if exprs and not parse_err:
+        tmp_path.write_text(_decode_gha_exprs(tmp_path.read_text(), exprs))
+    return not parse_err, output
+
+
+def _skip(path: Path, where: int, kind: str, output: str) -> None:
+    print(
+        f"  shfmt could not parse {kind} at {path}:{where + 1} — skipped",
+        file=sys.stderr,
+    )
+    print(f"    {output.strip()}", file=sys.stderr)
+
+
+def process_yaml_file(path: Path, tmp_path: Path) -> int:
+    text = path.read_text()
+    had_nl = text.endswith("\n")
+    lines = text.split("\n")
+    if had_nl:
+        lines = lines[:-1]
+    items = find_run_blocks(lines)
+    if not items:
+        return 0
+    changed = 0
+    # Process in reverse so earlier indices remain valid as we splice.
+    for item in reversed(items):
+        if isinstance(item, BlockRun):
+            body = lines[item.body_start : item.body_end]
+            tmp_path.write_text("\n".join(dedent(body, item.body_indent)) + "\n")
+            ok, output = shfmt_via_hook(tmp_path)
+            if not ok:
+                _skip(path, item.body_start, "block", output)
+                continue
+            formatted = tmp_path.read_text().rstrip("\n")
+            new_body = reindent(formatted.split("\n"), item.body_indent)
+            if new_body != body:
+                lines[item.body_start : item.body_end] = new_body
+                changed += 1
+        else:
+            tmp_path.write_text(item.value + "\n")
+            ok, output = shfmt_via_hook(tmp_path)
+            if not ok:
+                _skip(path, item.line_idx, "inline run", output)
+                continue
+            formatted = tmp_path.read_text().rstrip("\n")
+            if formatted == item.value:
+                continue
+            formatted_lines = formatted.split("\n")
+            if len(formatted_lines) == 1:
+                lines[item.line_idx] = f"{item.prefix}run: {formatted}"
+            else:
+                body_indent = len(item.prefix) + 2
+                lines[item.line_idx : item.line_idx + 1] = [
+                    f"{item.prefix}run: |",
+                    *reindent(formatted_lines, body_indent),
+                ]
+            changed += 1
+    new_text = "\n".join(lines) + ("\n" if had_nl else "")
+    if new_text != text:
+        path.write_text(new_text)
+    return changed
+
+
+def process_md_file(path: Path, tmp_path: Path) -> int:
+    text = path.read_text()
+    had_nl = text.endswith("\n")
+    lines = text.split("\n")
+    if had_nl:
+        lines = lines[:-1]
+    blocks = find_md_bash_blocks(lines)
+    if not blocks:
+        return 0
+    changed = 0
+    for block in reversed(blocks):
+        body = lines[block.body_start : block.body_end]
+        tmp_path.write_text("\n".join(dedent(body, block.body_indent)) + "\n")
+        ok, output = shfmt_via_hook(tmp_path)
+        if not ok:
+            _skip(path, block.open_line_idx, "```bash block", output)
+            continue
+        formatted = tmp_path.read_text().rstrip("\n")
+        formatted_lines = formatted.split("\n") if formatted else []
+        new_body = reindent(formatted_lines, block.body_indent)
+        if new_body != body:
+            lines[block.body_start : block.body_end] = new_body
+            changed += 1
+    new_text = "\n".join(lines) + ("\n" if had_nl else "")
+    if new_text != text:
+        path.write_text(new_text)
+    return changed
+
+
+def process_file(path: Path, tmp_path: Path) -> int:
+    if path.suffix in (".yml", ".yaml"):
+        return process_yaml_file(path, tmp_path)
+    if path.suffix == ".md":
+        return process_md_file(path, tmp_path)
+    return 0
+
+
+def gather_files(argv: list[str]) -> list[Path]:
+    """Return YAML workflow/action files and markdown files that we should
+    process — either the paths in `argv` or, when `argv` is empty, every
+    such file in the repo (skipping `external/`)."""
+    if argv:
+        candidates: list[Path] = [
+            (REPO / a).resolve() if not Path(a).is_absolute() else Path(a) for a in argv
+        ]
+    else:
+        gh = REPO / ".github"
+        candidates = [
+            *gh.rglob("*.yml"),
+            *gh.rglob("*.yaml"),
+            *(
+                p
+                for p in REPO.rglob("*.md")
+                if "external" not in p.relative_to(REPO).parts
+            ),
+        ]
+    return sorted(
+        p
+        for p in candidates
+        if p.exists()
+        and (
+            (p.suffix in (".yml", ".yaml") and ".github" in p.parts)
+            or p.suffix == ".md"
+        )
+    )
+
+
+def main(argv: list[str]) -> int:
+    files = gather_files(argv)
+    if not files:
+        return 0
+    with tempfile.TemporaryDirectory(prefix="format-inline-bash-") as tmpdir:
+        tmp_path = Path(tmpdir) / "shfmt.sh"
+        total = 0
+        for f in files:
+            n = process_file(f, tmp_path)
+            if n:
+                print(f"{f.relative_to(REPO)}: reformatted {n} block(s)")
+                total += n
+        return 1 if total else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

.github/workflows/build-nix-image.yml

@@ -100,8 +100,8 @@ jobs:
 
       - name: Create multi-arch manifests
         run: |
-          for tag in $(jq -cr '.tags[]' <<< "$DOCKER_METADATA_OUTPUT_JSON"); do
-            docker buildx imagetools create -t "$tag" "${tag}-amd64" "${tag}-arm64"
+          for tag in $(jq -cr '.tags[]' <<<"$DOCKER_METADATA_OUTPUT_JSON"); do
+              docker buildx imagetools create -t "$tag" "${tag}-amd64" "${tag}-arm64"
           done
 
       - name: Inspect image

.github/workflows/check-pr-description.yml

@@ -5,8 +5,17 @@ on:
     types:
       - checks_requested
   pull_request:
-    types: [opened, edited, reopened, synchronize, ready_for_review]
-    branches: [develop]
+    types:
+      - opened
+      - edited
+      - reopened
+      - synchronize
+      - ready_for_review
+    branches:
+      - develop
+      - "release-*"
+      - "release/*"
+      - "staging/*"
 
 jobs:
   check_description:
@@ -20,11 +29,11 @@ jobs:
         env:
           PR_BODY: ${{ github.event.pull_request.body }}
         if: ${{ github.event_name == 'pull_request' }}
-        run: printenv PR_BODY > pr_body.md
+        run: printenv PR_BODY >pr_body.md
 
       - name: Check PR description differs from template
         if: ${{ github.event_name == 'pull_request' }}
-        run: >
-          python .github/scripts/check-pr-description.py
-          --template-file .github/pull_request_template.md
-          --pr-body-file pr_body.md
+        run: |
+          python .github/scripts/check-pr-description.py \
+              --template-file .github/pull_request_template.md \
+              --pr-body-file pr_body.md

.github/workflows/check-pr-title.yml

@@ -5,10 +5,19 @@ on:
     types:
       - checks_requested
   pull_request:
-    types: [opened, edited, reopened, synchronize, ready_for_review]
-    branches: [develop]
+    types:
+      - opened
+      - edited
+      - reopened
+      - synchronize
+      - ready_for_review
+    branches:
+      - develop
+      - "release-*"
+      - "release/*"
+      - "staging/*"
 
 jobs:
   check_title:
     if: ${{ github.event.pull_request.draft != true }}
-    uses: XRPLF/actions/.github/workflows/check-pr-title.yml@291206777251b4d493641b5afbdf7c23009d2988
+    uses: XRPLF/actions/.github/workflows/check-pr-title.yml@cba1f0891650baf1a9c88624dc2d72573be2eb81

.github/workflows/on-pr.yml

@@ -98,7 +98,7 @@ jobs:
           READY: ${{ contains(github.event.pull_request.labels.*.name, 'Ready to merge') }}
           MERGE: ${{ github.event_name == 'merge_group' }}
         run: |
-          echo "go=${{ (env.DRAFT != 'true' && env.READY == 'true') || env.FILES == 'true' || env.MERGE == 'true' }}" >> "${GITHUB_OUTPUT}"
+          echo "go=${{ (env.DRAFT != 'true' && env.READY == 'true') || env.FILES == 'true' || env.MERGE == 'true' }}" >>"${GITHUB_OUTPUT}"
           cat "${GITHUB_OUTPUT}"
     outputs:
       go: ${{ steps.go.outputs.go == 'true' }}
@@ -168,9 +168,9 @@ jobs:
           PR_URL: ${{ github.event.pull_request.html_url }}
         run: |
           gh api --method POST -H "Accept: application/vnd.github+json" -H "X-GitHub-Api-Version: 2022-11-28" \
-          /repos/xrplf/clio/dispatches -f "event_type=check_libxrpl" \
-          -F "client_payload[ref]=${{ needs.upload-recipe.outputs.recipe_ref }}" \
-          -F "client_payload[pr_url]=${PR_URL}"
+              /repos/xrplf/clio/dispatches -f "event_type=check_libxrpl" \
+              -F "client_payload[ref]=${{ needs.upload-recipe.outputs.recipe_ref }}" \
+              -F "client_payload[pr_url]=${PR_URL}"
 
   passed:
     if: failure() || cancelled()

.github/workflows/pre-commit.yml

@@ -14,7 +14,7 @@ on:
 jobs:
   # Call the workflow in the XRPLF/actions repo that runs the pre-commit hooks.
   run-hooks:
-    uses: XRPLF/actions/.github/workflows/pre-commit.yml@5e942d61bf32f7557a7c159cfac4712a687b3e3a
+    uses: XRPLF/actions/.github/workflows/pre-commit.yml@cba1f0891650baf1a9c88624dc2d72573be2eb81
     with:
       runs_on: ubuntu-latest
       container: '{ "image": "ghcr.io/xrplf/ci/tools-rippled-pre-commit:sha-41ec7c1" }'

.github/workflows/reusable-build-docker-image.yml

@@ -53,7 +53,7 @@ jobs:
         env:
           PLATFORM: ${{ inputs.platform }}
         run: |
-          echo "arch=${PLATFORM##*/}" >> $GITHUB_OUTPUT
+          echo "arch=${PLATFORM##*/}" >>$GITHUB_OUTPUT
 
       - name: Set up Docker Buildx
         uses: docker/setup-buildx-action@d7f5e7f509e45cec5c76c4d5afdd7de93d0b3df5 # v4.1.0

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

@@ -113,7 +113,7 @@ jobs:
 
       - name: Set ccache log file
         if: ${{ inputs.ccache_enabled && runner.debug == '1' }}
-        run: echo "CCACHE_LOGFILE=${{ runner.temp }}/ccache.log" >> "${GITHUB_ENV}"
+        run: echo "CCACHE_LOGFILE=${{ runner.temp }}/ccache.log" >>"${GITHUB_ENV}"
 
       - name: Print build environment
         uses: XRPLF/actions/print-build-env@59dec886e4afb05a1724443af08baccbc045b574
@@ -146,11 +146,11 @@ jobs:
           CMAKE_ARGS: ${{ inputs.cmake_args }}
         run: |
           cmake \
-            -G '${{ runner.os == 'Windows' && 'Visual Studio 17 2022' || 'Ninja' }}' \
-            -DCMAKE_TOOLCHAIN_FILE:FILEPATH=build/generators/conan_toolchain.cmake \
-            -DCMAKE_BUILD_TYPE="${BUILD_TYPE}" \
-            ${CMAKE_ARGS} \
-            ..
+              -G '${{ runner.os == 'Windows' && 'Visual Studio 17 2022' || 'Ninja' }}' \
+              -DCMAKE_TOOLCHAIN_FILE:FILEPATH=build/generators/conan_toolchain.cmake \
+              -DCMAKE_BUILD_TYPE="${BUILD_TYPE}" \
+              ${CMAKE_ARGS} \
+              ..
 
       - name: Check protocol autogen files are up-to-date
         working-directory: ${{ env.BUILD_DIR }}
@@ -172,10 +172,10 @@ jobs:
           cmake --build . --target code_gen
           DIFF=$(git -C .. status --porcelain -- include/xrpl/protocol_autogen src/tests/libxrpl/protocol_autogen)
           if [ -n "${DIFF}" ]; then
-            echo "::error::Generated protocol files are out of date"
-            git -C .. diff -- include/xrpl/protocol_autogen src/tests/libxrpl/protocol_autogen
-            echo "${MESSAGE}"
-            exit 1
+              echo "::error::Generated protocol files are out of date"
+              git -C .. diff -- include/xrpl/protocol_autogen src/tests/libxrpl/protocol_autogen
+              echo "${MESSAGE}"
+              exit 1
           fi
 
       - name: Build the binary
@@ -186,18 +186,18 @@ jobs:
           CMAKE_TARGET: ${{ inputs.cmake_target }}
         run: |
           cmake \
-            --build . \
-            --config "${BUILD_TYPE}" \
-            --parallel "${BUILD_NPROC}" \
-            --target "${CMAKE_TARGET}"
+              --build . \
+              --config "${BUILD_TYPE}" \
+              --parallel "${BUILD_NPROC}" \
+              --target "${CMAKE_TARGET}"
 
       - name: Show ccache statistics
         if: ${{ inputs.ccache_enabled }}
         run: |
           ccache --show-stats -vv
           if [ '${{ runner.debug }}' = '1' ]; then
-            cat "${CCACHE_LOGFILE}"
-            curl ${CCACHE_REMOTE_STORAGE%|*}/status || true
+              cat "${CCACHE_LOGFILE}"
+              curl ${CCACHE_REMOTE_STORAGE%|*}/status || true
           fi
 
       - name: Upload the binary (Linux)
@@ -214,7 +214,7 @@ jobs:
         working-directory: ${{ env.BUILD_DIR }}
         run: |
           set -o pipefail
-          ./xrpld --definitions | python3 -m json.tool > server_definitions.json
+          ./xrpld --definitions | python3 -m json.tool >server_definitions.json
 
       - name: Upload server definitions
         if: ${{ github.event.repository.visibility == 'public' && inputs.config_name == 'debian-bookworm-gcc-13-amd64-release' }}
@@ -231,10 +231,10 @@ jobs:
         run: |
           ldd ./xrpld
           if [ "$(ldd ./xrpld | grep -E '(libstdc\+\+|libgcc)' | wc -l)" -eq 0 ]; then
-            echo 'The binary is statically linked.'
+              echo 'The binary is statically linked.'
           else
-            echo 'The binary is dynamically linked.'
-            exit 1
+              echo 'The binary is dynamically linked.'
+              exit 1
           fi
 
       - name: Verify presence of instrumentation (Linux)
@@ -250,12 +250,12 @@ jobs:
         run: |
           ASAN_OPTS="include=${GITHUB_WORKSPACE}/sanitizers/suppressions/runtime-asan-options.txt:suppressions=${GITHUB_WORKSPACE}/sanitizers/suppressions/asan.supp"
           if [[ "${CONFIG_NAME}" == *gcc* ]]; then
-            ASAN_OPTS="${ASAN_OPTS}:alloc_dealloc_mismatch=0"
+              ASAN_OPTS="${ASAN_OPTS}:alloc_dealloc_mismatch=0"
           fi
-          echo "ASAN_OPTIONS=${ASAN_OPTS}" >> ${GITHUB_ENV}
-          echo "TSAN_OPTIONS=include=${GITHUB_WORKSPACE}/sanitizers/suppressions/runtime-tsan-options.txt:suppressions=${GITHUB_WORKSPACE}/sanitizers/suppressions/tsan.supp" >> ${GITHUB_ENV}
-          echo "UBSAN_OPTIONS=include=${GITHUB_WORKSPACE}/sanitizers/suppressions/runtime-ubsan-options.txt:suppressions=${GITHUB_WORKSPACE}/sanitizers/suppressions/ubsan.supp" >> ${GITHUB_ENV}
-          echo "LSAN_OPTIONS=include=${GITHUB_WORKSPACE}/sanitizers/suppressions/runtime-lsan-options.txt:suppressions=${GITHUB_WORKSPACE}/sanitizers/suppressions/lsan.supp" >> ${GITHUB_ENV}
+          echo "ASAN_OPTIONS=${ASAN_OPTS}" >>${GITHUB_ENV}
+          echo "TSAN_OPTIONS=include=${GITHUB_WORKSPACE}/sanitizers/suppressions/runtime-tsan-options.txt:suppressions=${GITHUB_WORKSPACE}/sanitizers/suppressions/tsan.supp" >>${GITHUB_ENV}
+          echo "UBSAN_OPTIONS=include=${GITHUB_WORKSPACE}/sanitizers/suppressions/runtime-ubsan-options.txt:suppressions=${GITHUB_WORKSPACE}/sanitizers/suppressions/ubsan.supp" >>${GITHUB_ENV}
+          echo "LSAN_OPTIONS=include=${GITHUB_WORKSPACE}/sanitizers/suppressions/runtime-lsan-options.txt:suppressions=${GITHUB_WORKSPACE}/sanitizers/suppressions/lsan.supp" >>${GITHUB_ENV}
 
       - name: Run the separate tests
         if: ${{ !inputs.build_only }}
@@ -266,9 +266,9 @@ jobs:
           PARALLELISM: ${{ runner.os == 'Windows' && '1' || steps.nproc.outputs.nproc }}
         run: |
           ctest \
-            --output-on-failure \
-            -C "${BUILD_TYPE}" \
-            -j "${PARALLELISM}"
+              --output-on-failure \
+              -C "${BUILD_TYPE}" \
+              -j "${PARALLELISM}"
 
       - name: Run the embedded tests
         if: ${{ !inputs.build_only }}
@@ -278,7 +278,7 @@ jobs:
         run: |
           set -o pipefail
           # Coverage builds are slower due to instrumentation; use fewer parallel jobs to avoid flakiness
-          [ "$COVERAGE_ENABLED" = "true" ] && BUILD_NPROC=$(( BUILD_NPROC - 2 ))
+          [ "$COVERAGE_ENABLED" = "true" ] && BUILD_NPROC=$((BUILD_NPROC - 2))
           ./xrpld --unittest --unittest-jobs "${BUILD_NPROC}" 2>&1 | tee unittest.log
 
       - name: Show test failure summary
@@ -287,19 +287,19 @@ jobs:
           WORKING_DIR: ${{ runner.os == 'Windows' && format('{0}\{1}', env.BUILD_DIR, inputs.build_type) || env.BUILD_DIR }}
         run: |
           if [ ! -d "${WORKING_DIR}" ]; then
-            echo "Working directory '${WORKING_DIR}' does not exist."
-            exit 0
+              echo "Working directory '${WORKING_DIR}' does not exist."
+              exit 0
           fi
 
           cd "${WORKING_DIR}"
 
           if [ ! -f unittest.log ]; then
-            echo "unittest.log not found; embedded tests may not have run."
-            exit 0
+              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."
+              echo "Log present but no failure lines found in unittest.log."
           fi
       - name: Debug failure (Linux)
         if: ${{ failure() && runner.os == 'Linux' && !inputs.build_only }}
@@ -317,10 +317,10 @@ jobs:
           BUILD_TYPE: ${{ inputs.build_type }}
         run: |
           cmake \
-            --build . \
-            --config "${BUILD_TYPE}" \
-            --parallel "${BUILD_NPROC}" \
-            --target coverage
+              --build . \
+              --config "${BUILD_TYPE}" \
+              --parallel "${BUILD_NPROC}" \
+              --target coverage
 
       - name: Upload coverage report
         if: ${{ github.repository == 'XRPLF/rippled' && !inputs.build_only && env.COVERAGE_ENABLED == 'true' }}

.github/workflows/reusable-check-levelization.yml

@@ -38,9 +38,9 @@ jobs:
         run: |
           DIFF=$(git status --porcelain)
           if [ -n "${DIFF}" ]; then
-            # Print the differences to give the contributor a hint about what to
-            # expect when running levelization on their own machine.
-            git diff
-            echo "${MESSAGE}"
-            exit 1
+              # Print the differences to give the contributor a hint about what to
+              # expect when running levelization on their own machine.
+              git diff
+              echo "${MESSAGE}"
+              exit 1
           fi

.github/workflows/reusable-check-rename.yml

@@ -48,9 +48,9 @@ jobs:
         run: |
           DIFF=$(git status --porcelain)
           if [ -n "${DIFF}" ]; then
-            # Print the differences to give the contributor a hint about what to
-            # expect when running the renaming scripts on their own machine.
-            git diff
-            echo "${MESSAGE}"
-            exit 1
+              # Print the differences to give the contributor a hint about what to
+              # expect when running the renaming scripts on their own machine.
+              git diff
+              echo "${MESSAGE}"
+              exit 1
           fi

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

@@ -70,13 +70,13 @@ jobs:
         working-directory: ${{ env.BUILD_DIR }}
         run: |
           cmake \
-            -G 'Ninja' \
-            -DCMAKE_TOOLCHAIN_FILE:FILEPATH=build/generators/conan_toolchain.cmake \
-            -DCMAKE_BUILD_TYPE="${BUILD_TYPE}" \
-            -Dtests=ON \
-            -Dwerr=ON \
-            -Dxrpld=ON \
-            ..
+              -G 'Ninja' \
+              -DCMAKE_TOOLCHAIN_FILE:FILEPATH=build/generators/conan_toolchain.cmake \
+              -DCMAKE_BUILD_TYPE="${BUILD_TYPE}" \
+              -Dtests=ON \
+              -Dwerr=ON \
+              -Dxrpld=ON \
+              ..
 
       # clang-tidy needs headers generated from proto files
       - name: Build libxrpl.libpb
@@ -133,7 +133,7 @@ jobs:
       - name: Write issue header
         if: ${{ steps.run_clang_tidy.outcome != 'success' }}
         run: |
-          cat > "${ISSUE_FILE}" <<EOF
+          cat >"${ISSUE_FILE}" <<EOF
           ## Clang-tidy Check Failed
 
           ### Clang-tidy Output:
@@ -144,30 +144,30 @@ jobs:
         if: ${{ steps.run_clang_tidy.outcome != 'success' }}
         run: |
           if [ -f "${OUTPUT_FILE}" ]; then
-            # Extract lines containing 'error:', 'warning:', or 'note:'
-            grep -E '(error:|warning:|note:)' "${OUTPUT_FILE}" > filtered-output.txt || true
+              # Extract lines containing 'error:', 'warning:', or 'note:'
+              grep -E '(error:|warning:|note:)' "${OUTPUT_FILE}" >filtered-output.txt || true
 
-            # If filtered output is empty, use original (might be a different error format)
-            if [ ! -s filtered-output.txt ]; then
-              cp "${OUTPUT_FILE}" filtered-output.txt
-            fi
+              # If filtered output is empty, use original (might be a different error format)
+              if [ ! -s filtered-output.txt ]; then
+                  cp "${OUTPUT_FILE}" filtered-output.txt
+              fi
 
-            # Truncate if too large
-            head -c 60000 filtered-output.txt >> "${ISSUE_FILE}"
-            if [ "$(wc -c < filtered-output.txt)" -gt 60000 ]; then
-              echo "" >> "${ISSUE_FILE}"
-              echo "... (output truncated, see artifacts for full output)" >> "${ISSUE_FILE}"
-            fi
+              # Truncate if too large
+              head -c 60000 filtered-output.txt >>"${ISSUE_FILE}"
+              if [ "$(wc -c <filtered-output.txt)" -gt 60000 ]; then
+                  echo "" >>"${ISSUE_FILE}"
+                  echo "... (output truncated, see artifacts for full output)" >>"${ISSUE_FILE}"
+              fi
 
-            rm filtered-output.txt
+              rm filtered-output.txt
           else
-            echo "No output file found" >> "${ISSUE_FILE}"
+              echo "No output file found" >>"${ISSUE_FILE}"
           fi
 
       - name: Append issue footer
         if: ${{ steps.run_clang_tidy.outcome != 'success' }}
         run: |
-          cat >> "${ISSUE_FILE}" <<EOF
+          cat >>"${ISSUE_FILE}" <<EOF
           \`\`\`
 
           ---
@@ -176,7 +176,7 @@ jobs:
 
       - name: Create issue
         if: ${{ steps.run_clang_tidy.outcome != 'success' && inputs.create_issue_on_failure }}
-        uses: XRPLF/actions/create-issue@36d450d12d301e8410c1b7936e5de70c291cbe36
+        uses: XRPLF/actions/create-issue@2b8bc36af85b88bca0dd7bfac2e2dc05f94ad712
         with:
           title: "Clang-tidy check failed"
           body_file: ${{ env.ISSUE_FILE }}

.github/workflows/reusable-package.yml

@@ -39,7 +39,7 @@ jobs:
         id: generate
         working-directory: .github/scripts/strategy-matrix
         run: |
-          ./generate.py --packaging --config=linux.json >> "${GITHUB_OUTPUT}"
+          ./generate.py --packaging --config=linux.json >>"${GITHUB_OUTPUT}"
 
   generate-version:
     runs-on: ubuntu-latest

.github/workflows/reusable-strategy-matrix.yml

@@ -42,4 +42,4 @@ jobs:
         env:
           GENERATE_CONFIG: ${{ inputs.os != '' && format('--config={0}.json', inputs.os) || '' }}
           GENERATE_OPTION: ${{ inputs.strategy_matrix == 'all' && '--all' || '' }}
-        run: ./generate.py ${GENERATE_OPTION} ${GENERATE_CONFIG} >> "${GITHUB_OUTPUT}"
+        run: ./generate.py ${GENERATE_OPTION} ${GENERATE_CONFIG} >>"${GITHUB_OUTPUT}"

.gitignore

@@ -86,3 +86,5 @@ __pycache__
 
 # clangd cache
 /.cache
+labrun/
+build-release/

.pre-commit-config.yaml

@@ -66,6 +66,19 @@ repos:
       - id: shfmt
         args: [--write, --indent=4, --case-indent=true]
 
+  - repo: local
+    hooks:
+      - id: format-inline-bash-workflows
+        name: "format `run:` blocks in workflows/actions"
+        entry: ./.github/scripts/format-inline-bash.py
+        language: python
+        files: ^\.github/(workflows|actions)/.*\.ya?ml$
+      - id: format-inline-bash-markdown
+        name: "format ```bash blocks in markdown"
+        entry: ./.github/scripts/format-inline-bash.py
+        language: python
+        files: \.md$
+
   - repo: https://github.com/streetsidesoftware/cspell-cli
     rev: 4643f154907327ee0a2c7038f0296e0dd77d9776 # frozen: v10.0.0
     hooks:

BUILD.md

@@ -151,8 +151,8 @@ git init
 git remote add origin git@github.com:XRPLF/conan-center-index.git
 git sparse-checkout init
 for recipe in "${recipes[@]}"; do
-  echo "Checking out recipe '${recipe}'..."
-  git sparse-checkout add recipes/${recipe}
+    echo "Checking out recipe '${recipe}'..."
+    git sparse-checkout add recipes/${recipe}
 done
 git fetch origin master
 git checkout master
@@ -180,7 +180,7 @@ the new recipe will be automatically pulled from the official Conan Center.
 
 If you see an error similar to the following after running `conan profile show`:
 
-```bash
+```text
 ERROR: Invalid setting '17' is not a valid 'settings.compiler.version' value.
 Possible values are ['5.0', '5.1', '6.0', '6.1', '7.0', '7.3', '8.0', '8.1',
 '9.0', '9.1', '10.0', '11.0', '12.0', '13', '13.0', '13.1', '14', '14.0', '15',

cspell.config.yaml

@@ -93,6 +93,7 @@ words:
   - daria
   - dcmake
   - dearmor
+  - dedented
   - deleteme
   - demultiplexer
   - deserializaton

include/xrpl/ledger/detail/ApplyStateTable.h

@@ -4,11 +4,15 @@
 #include <xrpl/ledger/OpenView.h>
 #include <xrpl/ledger/RawView.h>
 #include <xrpl/ledger/ReadView.h>
+#include <xrpl/protocol/LedgerFormats.h>
 #include <xrpl/protocol/TER.h>
 #include <xrpl/protocol/TxMeta.h>
 #include <xrpl/protocol/XRPAmount.h>
 
 #include <memory>
+#ifndef NDEBUG
+#include <map>
+#endif
 
 namespace xrpl::detail {
 
@@ -103,7 +107,39 @@ public:
         return dropsDestroyed_;
     }
 
+#ifndef NDEBUG
+    /** Every ledger entry this table has read or written, mapped to its type.
+
+        Populated in DEBUG builds by the access methods below (reads via
+        read/exists/peek and writes via insert/update/replace/erase). Directory
+        iteration via succ() is deliberately NOT recorded — see AccessSet for
+        why directory entries are out of scope for conflict tracking. Used by
+        the parallel-apply access-set assertion to verify a transactor's
+        declared footprint is a superset of what it actually touched.
+    */
+    [[nodiscard]] std::map<key_type, LedgerEntryType> const&
+    touchedEntries() const
+    {
+        return touched_;
+    }
+#endif
+
 private:
+#ifndef NDEBUG
+    void
+    recordTouch(key_type const& key, LedgerEntryType type) const
+    {
+        // Prefer ltDIR_NODE on collision so directory pages stay identifiable
+        // regardless of access order; non-dir objects are never accessed via a
+        // directory keylet, so this never mislabels a real object.
+        auto const [it, inserted] = touched_.try_emplace(key, type);
+        if (!inserted && type == ltDIR_NODE)
+            it->second = ltDIR_NODE;
+    }
+
+    mutable std::map<key_type, LedgerEntryType> touched_;
+#endif
+
     using Mods = hash_map<key_type, std::shared_ptr<SLE>>;
 
     static void

include/xrpl/ledger/detail/ApplyViewBase.h

@@ -95,6 +95,16 @@ public:
     void
     rawDestroyXRP(XRPAmount const& feeDrops) override;
 
+#ifndef NDEBUG
+    /** Every ledger entry touched (read or written) by this view, with type.
+        DEBUG-only; backs the parallel-apply access-set assertion. */
+    [[nodiscard]] std::map<uint256, LedgerEntryType> const&
+    touchedEntries() const
+    {
+        return items_.touchedEntries();
+    }
+#endif
+
 protected:
     ApplyFlags flags_;
     ReadView const* base_;

include/xrpl/protocol_autogen/README.md

@@ -15,8 +15,8 @@ Generation requires a one-time setup step to create a virtual environment
 and install Python dependencies, followed by running the generation target:
 
 ```bash
-cmake --build . --target setup_code_gen  # create venv and install dependencies (once)
-cmake --build . --target code_gen        # generate code
+cmake --build . --target setup_code_gen # create venv and install dependencies (once)
+cmake --build . --target code_gen       # generate code
 ```
 
 By default, `CODEGEN_VENV_DIR` points to `.venv` in the project root. The

include/xrpl/tx/AccessSet.h

@@ -0,0 +1,92 @@
+#pragma once
+
+#include <xrpl/basics/base_uint.h>
+
+#include <boost/container/flat_set.hpp>
+
+#include <set>
+
+namespace xrpl {
+
+/** The statically-declared ledger footprint of a single transaction.
+
+    An AccessSet enumerates the ledger entries a transaction may read or write,
+    grouped by category for readability. It is produced by `accessSetOf()` from
+    the signed transaction body (and, where required, a read-only snapshot of
+    the closed ledger) — never from another transaction's in-flight writes.
+
+    Two transactions are independent (safe to apply concurrently) iff their
+    AccessSets do not conflict; see `conflictsWith()`. A transaction whose
+    footprint cannot be enumerated statically sets `touchesGlobal` and is
+    serialized against everything.
+
+    Directory-node (ltDIR_NODE) entries are intentionally NOT represented here.
+    Owner-directory bookkeeping is a consequence of mutating the owning account,
+    whose AccountRoot is already declared; shared (book) directories belong only
+    to transactors that are `touchesGlobal` in this version. Conflict detection
+    on directories is therefore subsumed by the account/object entries.
+
+    The category split is for human readability and future scheduler heuristics;
+    conflict detection itself operates on the flat union, `keys()`.
+*/
+struct AccessSet
+{
+    boost::container::flat_set<uint256> accounts;     // AccountRoot entries
+    boost::container::flat_set<uint256> trustlines;   // RippleState entries
+    boost::container::flat_set<uint256> offerBooks;   // book directory roots
+    boost::container::flat_set<uint256> ammPools;     // AMM root entries
+    boost::container::flat_set<uint256> nftPages;     // NFToken page roots
+    boost::container::flat_set<uint256> miscObjects;  // escrow/check/ticket/etc.
+
+    /** When true, the footprint is not statically known; serialize this tx. */
+    bool touchesGlobal = false;
+
+    /** The conservative "I touch everything" footprint. */
+    [[nodiscard]] static AccessSet
+    global()
+    {
+        AccessSet a;
+        a.touchesGlobal = true;
+        return a;
+    }
+
+    /** The flat union of every declared entry key, across all categories. */
+    [[nodiscard]] std::set<uint256>
+    keys() const
+    {
+        std::set<uint256> out;
+        for (auto const* s :
+             {&accounts, &trustlines, &offerBooks, &ammPools, &nftPages, &miscObjects})
+            out.insert(s->begin(), s->end());
+        return out;
+    }
+
+    /** True if these two transactions cannot safely apply concurrently.
+
+        Either side touching global state forces a conflict; otherwise the two
+        conflict iff their declared key sets intersect.
+    */
+    [[nodiscard]] bool
+    conflictsWith(AccessSet const& other) const
+    {
+        if (touchesGlobal || other.touchesGlobal)
+            return true;
+
+        auto const a = keys();
+        auto const b = other.keys();
+        auto i = a.begin();
+        auto j = b.begin();
+        while (i != a.end() && j != b.end())
+        {
+            if (*i < *j)
+                ++i;
+            else if (*j < *i)
+                ++j;
+            else
+                return true;
+        }
+        return false;
+    }
+};
+
+}  // namespace xrpl

include/xrpl/tx/ApplyContext.h

@@ -112,6 +112,24 @@ public:
     TER
     checkInvariants(TER const result, XRPAmount const fee);
 
+#ifndef NDEBUG
+    /** The read-only base (closed-ledger) snapshot this tx applies against. */
+    [[nodiscard]] ReadView const&
+    baseView() const
+    {
+        return base_;
+    }
+
+    /** Every ledger entry touched during apply, with type. Backs the
+        access-set assertion in Transactor::operator(). */
+    [[nodiscard]] std::map<uint256, LedgerEntryType> const&
+    touchedEntries() const
+    {
+        // NOLINTNEXTLINE(bugprone-unchecked-optional-access) view_ emplaced in ctor
+        return view_->touchedEntries();
+    }
+#endif
+
 private:
     static TER
     failInvariantCheck(TER const result);

include/xrpl/tx/Schedule.h

@@ -0,0 +1,128 @@
+#pragma once
+
+#include <xrpl/beast/utility/Journal.h>
+#include <xrpl/ledger/RawView.h>
+#include <xrpl/ledger/ReadView.h>
+#include <xrpl/protocol/STTx.h>
+#include <xrpl/tx/AccessSet.h>
+
+#include <cstddef>
+#include <memory>
+#include <vector>
+
+namespace xrpl {
+
+class ServiceRegistry;
+
+/** A maximal set of transactions that conflict (transitively) with each other.
+
+    The transactions are held in canonical (input) order, which is significant:
+    within a group they must be applied serially in this order. Two distinct
+    groups are independent by construction — their declared access sets are
+    disjoint — so groups may be applied concurrently, and the order in which
+    groups are applied does not affect the resulting state.
+*/
+struct ConflictGroup
+{
+    std::vector<std::shared_ptr<STTx const>> txns;
+};
+
+/** The partitioning of a canonically-ordered transaction set for parallel
+    application.
+
+    Either the set was partitioned into independent `groups` (the parallel
+    case), or scheduling fell back to a single serial ordering in `serial`
+    (the conservative case — see `scheduleApply`). Exactly one of the two is
+    populated; `fullySerial` says which.
+*/
+struct Schedule
+{
+    std::vector<ConflictGroup> groups;
+    std::vector<std::shared_ptr<STTx const>> serial;
+    bool fullySerial = false;
+
+    /** Total transactions scheduled, across groups or the serial list. */
+    [[nodiscard]] std::size_t
+    size() const
+    {
+        if (fullySerial)
+            return serial.size();
+        std::size_t n = 0;
+        for (auto const& g : groups)
+            n += g.txns.size();
+        return n;
+    }
+};
+
+/** Partition a canonically-ordered transaction set into independent groups by
+    static access-set conflict, for parallel application.
+
+    For each transaction, `accessSetOf` is consulted against `base` (the closed
+    -ledger snapshot the round applies to). Transactions are unioned into the
+    same group iff their access sets conflict — i.e. they share any declared
+    ledger entry, or (implicitly, via the access set) act on the same account.
+
+    If ANY transaction declares `touchesGlobal` (a pseudo-transaction such as
+    SetFee/EnableAmendment/UNLModify, or any not-yet-migrated transactor), the
+    schedule falls back to fully serial in canonical order. This is the
+    conservative flag-ledger handling: correctness over throughput, at a cost of
+    ~1/256 of ledgers. A later version may apply the global transactions first
+    and parallelize the remainder.
+
+    Deterministic: identical input yields an identical schedule (groups are
+    ordered by their lowest canonical index, transactions within a group keep
+    canonical order). Applies nothing and reads only `base`.
+*/
+Schedule
+scheduleApply(std::vector<std::shared_ptr<STTx const>> const& txns, ReadView const& base);
+
+/** Outcome of applyScheduled. */
+struct ScheduledApplyResult
+{
+    std::size_t groupCount = 0;  // independent groups (1 if fully serial)
+    std::size_t applied = 0;     // transactions that applied successfully
+    bool fullySerial = false;
+};
+
+/** Apply a canonically-ordered transaction set via its schedule.
+
+    Schedules `txns` (see `scheduleApply`), then applies each independent group
+    in its own `OpenView` layered over the immutable `closed` snapshot, and
+    merges the per-group write-sets into `to`. Because distinct groups have
+    disjoint access sets, their write-sets are disjoint and the merge is
+    order-independent — so the resulting state in `to` is byte-identical to a
+    serial canonical apply. (That equivalence is what the differential test
+    asserts, and it is the correctness contract a parallel executor relies on.)
+
+    `workers` controls concurrency of the per-group apply phase:
+      - `workers <= 1`: groups applied sequentially (deterministic baseline).
+      - `workers > 1`: up to `workers` groups apply concurrently, each in its own
+        view over the immutable `closed` snapshot; the write-sets are then merged
+        into `to` sequentially in deterministic group order.
+    The result is identical for any `workers` value (the merge is order-
+    independent because groups are disjoint, and is performed in a fixed order).
+
+    NOTE on the threaded path: it is correct by construction here, but it is NOT
+    certified for production consensus use. A non-deterministic apply forks the
+    network, so before the threaded path may be trusted it needs ThreadSanitizer
+    clean runs, adversarial scheduling (Antithesis), and a mainnet-history replay
+    differential — none of which a unit test establishes. The concurrent reads
+    of `closed` and the shared `registry` are the surfaces that must be cleared.
+
+    @param registry service registry used by the apply pipeline.
+    @param closed   the immutable closed-ledger snapshot to read and schedule against.
+    @param to       the target ledger receiving the merged writes.
+    @param txns     transactions in canonical order.
+    @param j        journal.
+    @param workers  max concurrent group-apply threads (default 1 = sequential).
+*/
+ScheduledApplyResult
+applyScheduled(
+    ServiceRegistry& registry,
+    ReadView const& closed,
+    TxsRawView& to,
+    std::vector<std::shared_ptr<STTx const>> const& txns,
+    beast::Journal j,
+    unsigned workers = 1);
+
+}  // namespace xrpl

include/xrpl/tx/Transactor.h

@@ -4,6 +4,7 @@
 #include <xrpl/beast/utility/WrappedSink.h>
 #include <xrpl/protocol/Permissions.h>
 #include <xrpl/protocol/XRPAmount.h>
+#include <xrpl/tx/AccessSet.h>
 #include <xrpl/tx/ApplyContext.h>
 #include <xrpl/tx/applySteps.h>
 
@@ -222,10 +223,39 @@ public:
         return tesSUCCESS;
     }
 
+    /** The static ledger footprint this transaction may read or write.
+
+        Consumed by the parallel-apply scheduler to decide which transactions
+        can apply concurrently. The base implementation is fail-safe: it
+        declares `touchesGlobal`, so an un-migrated transactor is serialized
+        against everything. A transactor opts into concurrency by hiding this
+        with its own static `accessSetOf` that enumerates exactly what it
+        touches. Like preclaim/calculateBaseFee, this is dispatched by
+        compile-time name hiding, not virtual dispatch.
+
+        @param tx   the signed transaction.
+        @param base a read-only snapshot of the ledger the tx applies to, for
+                    the few transactors whose footprint needs a state lookup
+                    (e.g. resolving an AMM pseudo-account). Most ignore it.
+    */
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base)
+    {
+        return AccessSet::global();
+    }
+
     static NotTEC
     checkPermission(ReadView const& view, STTx const& tx);
     /////////////////////////////////////////////////////
 
+    /** The ledger footprint every transaction incurs, regardless of type: the
+        actor's AccountRoot, the fee-payer's AccountRoot (when delegated), and
+        the consumed Ticket object (when ticket-sequenced). Directory pages are
+        excluded by design — see AccessSet. A migrated `accessSetOf` seeds its
+        result with this and adds its type-specific entries. */
+    static AccessSet
+    commonAccountFootprint(STTx const& tx);
+
     // Interface used by AccountDelete
     static TER
     ticketDelete(
@@ -380,6 +410,15 @@ private:
 
     void trapTransaction(uint256) const;
 
+#ifndef NDEBUG
+    /** DEBUG: assert the transaction's actual ledger footprint is a subset of
+        the access set declared by `accessSetOf`, or — for `touchesGlobal`
+        transactors — optionally record the measured footprint for the
+        hard-transactor audit. Called only on a clean tesSUCCESS apply. */
+    void
+    verifyAccessSet() const;
+#endif
+
     /** Performs early sanity checks on the account and fee fields.
 
         (And passes flagMask to preflight0)

include/xrpl/tx/applySteps.h

@@ -2,6 +2,7 @@
 
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/ledger/ApplyViewImpl.h>
+#include <xrpl/tx/AccessSet.h>
 
 namespace xrpl {
 
@@ -308,6 +309,19 @@ preclaim(PreflightResult const& preflightResult, ServiceRegistry& registry, Open
 XRPAmount
 calculateBaseFee(ReadView const& view, STTx const& tx);
 
+/** Compute the static ledger footprint of a transaction.
+
+    Dispatches to the concrete transactor's `accessSetOf`. Un-migrated
+    transactors return `touchesGlobal`. Used by the parallel-apply scheduler
+    (and the DEBUG access-set assertion) to reason about which transactions can
+    apply concurrently. No validation is implied.
+
+    @param tx   the transaction.
+    @param base a read-only snapshot of the ledger the tx applies to.
+*/
+AccessSet
+accessSetOf(STTx const& tx, ReadView const& base);
+
 /** Return the minimum fee that an "ordinary" transaction would pay.
 
     When computing the FeeLevel for a transaction the TxQ sometimes needs

include/xrpl/tx/transactors/account/AccountSet.h

@@ -29,6 +29,9 @@ public:
     static TER
     preclaim(PreclaimContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
 

include/xrpl/tx/transactors/account/SetRegularKey.h

@@ -19,6 +19,9 @@ public:
     static XRPAmount
     calculateBaseFee(ReadView const& view, STTx const& tx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
 

include/xrpl/tx/transactors/account/SignerListSet.h

@@ -36,6 +36,9 @@ public:
     static NotTEC
     preflight(PreflightContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
     void

include/xrpl/tx/transactors/delegate/DelegateSet.h

@@ -19,6 +19,9 @@ public:
     static TER
     preclaim(PreclaimContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
 

include/xrpl/tx/transactors/did/DIDDelete.h

@@ -22,6 +22,9 @@ public:
     static TER
     deleteSLE(ApplyView& view, std::shared_ptr<SLE> sle, AccountID const owner, beast::Journal j);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
 

include/xrpl/tx/transactors/did/DIDSet.h

@@ -16,6 +16,9 @@ public:
     static NotTEC
     preflight(PreflightContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
 

include/xrpl/tx/transactors/oracle/OracleDelete.h

@@ -28,6 +28,9 @@ public:
     static TER
     preclaim(PreclaimContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
 

include/xrpl/tx/transactors/oracle/OracleSet.h

@@ -28,6 +28,9 @@ public:
     static TER
     preclaim(PreclaimContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
 

include/xrpl/tx/transactors/payment/DepositPreauth.h

@@ -22,6 +22,9 @@ public:
     static TER
     preclaim(PreclaimContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
 

include/xrpl/tx/transactors/payment/Payment.h

@@ -37,6 +37,9 @@ public:
     static TER
     preclaim(PreclaimContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
 

include/xrpl/tx/transactors/permissioned_domain/PermissionedDomainDelete.h

@@ -19,6 +19,9 @@ public:
     static TER
     preclaim(PreclaimContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     /** Attempt to delete the Permissioned Domain. */
     TER
     doApply() override;

include/xrpl/tx/transactors/system/TicketCreate.h

@@ -56,6 +56,9 @@ public:
     static TER
     preclaim(PreclaimContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     /** Precondition: fee collection is likely.  Attempt to create ticket(s). */
     TER
     doApply() override;

include/xrpl/tx/transactors/token/TrustSet.h

@@ -26,6 +26,9 @@ public:
     static TER
     preclaim(PreclaimContext const& ctx);
 
+    static AccessSet
+    accessSetOf(STTx const& tx, ReadView const& base);
+
     TER
     doApply() override;
 

package/README.md

@@ -74,10 +74,10 @@ VERSION=2.4.0-local
 PKG_RELEASE=1
 
 docker run --rm \
-  -v "$(pwd):/src" \
-  -w /src \
-  "$IMAGE" \
-  ./package/build_pkg.sh --pkg-version "$VERSION" --pkg-release "$PKG_RELEASE"
+    -v "$(pwd):/src" \
+    -w /src \
+    "$IMAGE" \
+    ./package/build_pkg.sh --pkg-version "$VERSION" --pkg-release "$PKG_RELEASE"
 
 # Output:
 #   build/debbuild/*.deb         (DEB + dbgsym .ddeb)
@@ -92,12 +92,12 @@ needed, but the host toolchain replaces the pinned CI image:
 
 ```bash
 cmake \
-  -Dxrpld=ON \
-  -Dxrpld_version=2.4.0-local \
-  -Dtests=OFF \
-  ..
+    -Dxrpld=ON \
+    -Dxrpld_version=2.4.0-local \
+    -Dtests=OFF \
+    ..
 
-cmake --build . --target package       # deb on Debian/Ubuntu, rpm on RHEL
+cmake --build . --target package # deb on Debian/Ubuntu, rpm on RHEL
 ```
 
 The `cmake/XrplPackaging.cmake` module defines the target only if at least one

src/libxrpl/ledger/ApplyStateTable.cpp

@@ -287,6 +287,9 @@ ApplyStateTable::apply(
 bool
 ApplyStateTable::exists(ReadView const& base, Keylet const& k) const
 {
+#ifndef NDEBUG
+    recordTouch(k.key, k.type);
+#endif
     auto const iter = items_.find(k.key);
     if (iter == items_.end())
         return base.exists(k);
@@ -342,6 +345,11 @@ ApplyStateTable::succ(
 std::shared_ptr<SLE const>
 ApplyStateTable::read(ReadView const& base, Keylet const& k) const
 {
+#ifndef NDEBUG
+    // Record even on a nullptr result: an absence probe still conflicts with a
+    // concurrent transaction that would create the key.
+    recordTouch(k.key, k.type);
+#endif
     auto const iter = items_.find(k.key);
     if (iter == items_.end())
         return base.read(k);
@@ -364,6 +372,9 @@ ApplyStateTable::read(ReadView const& base, Keylet const& k) const
 std::shared_ptr<SLE>
 ApplyStateTable::peek(ReadView const& base, Keylet const& k)
 {
+#ifndef NDEBUG
+    recordTouch(k.key, k.type);
+#endif
     auto iter = items_.lower_bound(k.key);
     if (iter == items_.end() || iter->first != k.key)
     {
@@ -398,6 +409,9 @@ ApplyStateTable::peek(ReadView const& base, Keylet const& k)
 void
 ApplyStateTable::erase(ReadView const& base, std::shared_ptr<SLE> const& sle)
 {
+#ifndef NDEBUG
+    recordTouch(sle->key(), sle->getType());
+#endif
     auto const iter = items_.find(sle->key());
     if (iter == items_.end())
         Throw<std::logic_error>("ApplyStateTable::erase: missing key");
@@ -422,6 +436,9 @@ ApplyStateTable::erase(ReadView const& base, std::shared_ptr<SLE> const& sle)
 void
 ApplyStateTable::rawErase(ReadView const& base, std::shared_ptr<SLE> const& sle)
 {
+#ifndef NDEBUG
+    recordTouch(sle->key(), sle->getType());
+#endif
     using namespace std;
     auto const result = items_.emplace(
         piecewise_construct, forward_as_tuple(sle->key()), forward_as_tuple(Action::Erase, sle));
@@ -447,6 +464,9 @@ ApplyStateTable::rawErase(ReadView const& base, std::shared_ptr<SLE> const& sle)
 void
 ApplyStateTable::insert(ReadView const& base, std::shared_ptr<SLE> const& sle)
 {
+#ifndef NDEBUG
+    recordTouch(sle->key(), sle->getType());
+#endif
     auto const iter = items_.lower_bound(sle->key());
     if (iter == items_.end() || iter->first != sle->key())
     {
@@ -477,6 +497,9 @@ ApplyStateTable::insert(ReadView const& base, std::shared_ptr<SLE> const& sle)
 void
 ApplyStateTable::replace(ReadView const& base, std::shared_ptr<SLE> const& sle)
 {
+#ifndef NDEBUG
+    recordTouch(sle->key(), sle->getType());
+#endif
     auto const iter = items_.lower_bound(sle->key());
     if (iter == items_.end() || iter->first != sle->key())
     {
@@ -506,6 +529,9 @@ ApplyStateTable::replace(ReadView const& base, std::shared_ptr<SLE> const& sle)
 void
 ApplyStateTable::update(ReadView const& base, std::shared_ptr<SLE> const& sle)
 {
+#ifndef NDEBUG
+    recordTouch(sle->key(), sle->getType());
+#endif
     auto const iter = items_.find(sle->key());
     if (iter == items_.end())
         Throw<std::logic_error>("ApplyStateTable::update: missing key");

src/libxrpl/tx/Schedule.cpp

@@ -0,0 +1,217 @@
+#include <xrpl/tx/Schedule.h>
+
+#include <xrpl/basics/base_uint.h>
+#include <xrpl/beast/hash/uhash.h>
+#include <xrpl/ledger/ApplyView.h>
+#include <xrpl/ledger/OpenView.h>
+#include <xrpl/tx/apply.h>
+#include <xrpl/tx/applySteps.h>
+
+#include <algorithm>
+#include <atomic>
+#include <cstddef>
+#include <map>
+#include <numeric>
+#include <optional>
+#include <thread>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+namespace xrpl {
+
+namespace {
+
+// Disjoint-set (union-find) with path compression + union by size.
+class UnionFind
+{
+public:
+    explicit UnionFind(std::size_t n) : parent_(n), size_(n, 1)
+    {
+        std::iota(parent_.begin(), parent_.end(), std::size_t{0});
+    }
+
+    std::size_t
+    find(std::size_t x)
+    {
+        while (parent_[x] != x)
+            x = parent_[x] = parent_[parent_[x]];
+        return x;
+    }
+
+    void
+    unite(std::size_t a, std::size_t b)
+    {
+        a = find(a);
+        b = find(b);
+        if (a == b)
+            return;
+        if (size_[a] < size_[b])
+            std::swap(a, b);
+        parent_[b] = a;
+        size_[a] += size_[b];
+    }
+
+private:
+    std::vector<std::size_t> parent_;
+    std::vector<std::size_t> size_;
+};
+
+}  // namespace
+
+Schedule
+scheduleApply(std::vector<std::shared_ptr<STTx const>> const& txns, ReadView const& base)
+{
+    Schedule sched;
+
+    std::vector<AccessSet> access;
+    access.reserve(txns.size());
+    bool anyGlobal = false;
+    for (auto const& tx : txns)
+    {
+        access.push_back(accessSetOf(*tx, base));
+        anyGlobal = anyGlobal || access.back().touchesGlobal;
+    }
+
+    // Conservative fallback: any global-touching (or pseudo) transaction forces
+    // a single serial ordering. Correctness over throughput on flag ledgers.
+    if (anyGlobal)
+    {
+        sched.fullySerial = true;
+        sched.serial = txns;
+        return sched;
+    }
+
+    // Union transactions that share any declared ledger entry. An inverted
+    // index (key -> first transaction seen touching it) makes this near-linear
+    // in the total number of declared keys, rather than O(n^2) pairwise.
+    UnionFind uf(txns.size());
+    std::unordered_map<uint256, std::size_t, beast::Uhash<>> ownerOfKey;
+    for (std::size_t i = 0; i < access.size(); ++i)
+    {
+        for (auto const& key : access[i].keys())
+        {
+            auto const [it, inserted] = ownerOfKey.try_emplace(key, i);
+            if (!inserted)
+                uf.unite(i, it->second);
+        }
+    }
+
+    // Collect components. Iterating i ascending preserves canonical order within
+    // each group; groups are then ordered deterministically by the transaction
+    // id of their first (lowest canonical index) member.
+    std::map<std::size_t, std::vector<std::size_t>> components;
+    for (std::size_t i = 0; i < txns.size(); ++i)
+        components[uf.find(i)].push_back(i);
+
+    sched.groups.reserve(components.size());
+    for (auto const& [root, members] : components)
+    {
+        ConflictGroup group;
+        group.txns.reserve(members.size());
+        for (auto const idx : members)
+            group.txns.push_back(txns[idx]);
+        sched.groups.push_back(std::move(group));
+    }
+
+    // Deterministic group order: by each group's first (lowest-index) member.
+    std::sort(
+        sched.groups.begin(),
+        sched.groups.end(),
+        [](ConflictGroup const& a, ConflictGroup const& b) {
+            return a.txns.front()->getTransactionID() < b.txns.front()->getTransactionID();
+        });
+
+    return sched;
+}
+
+namespace {
+
+// Apply a group's transactions into a fresh isolated view over `closed`,
+// returning the view (with its accumulated, not-yet-merged write-set) and the
+// count that applied. Reads only `closed`, so this is independent of every
+// other group.
+std::pair<OpenView, std::size_t>
+applyGroupIsolated(
+    ServiceRegistry& registry,
+    ReadView const& closed,
+    std::vector<std::shared_ptr<STTx const>> const& group,
+    beast::Journal j)
+{
+    OpenView gv(&closed);
+    std::size_t applied = 0;
+    for (auto const& tx : group)
+    {
+        if (apply(registry, gv, *tx, TapNone, j).applied)
+            ++applied;
+    }
+    return {std::move(gv), applied};
+}
+
+}  // namespace
+
+ScheduledApplyResult
+applyScheduled(
+    ServiceRegistry& registry,
+    ReadView const& closed,
+    TxsRawView& to,
+    std::vector<std::shared_ptr<STTx const>> const& txns,
+    beast::Journal j,
+    unsigned workers)
+{
+    auto const sched = scheduleApply(txns, closed);
+
+    ScheduledApplyResult res;
+    res.fullySerial = sched.fullySerial;
+
+    if (sched.fullySerial)
+    {
+        res.groupCount = 1;
+        auto [gv, applied] = applyGroupIsolated(registry, closed, sched.serial, j);
+        gv.apply(to);
+        res.applied = applied;
+        return res;
+    }
+
+    res.groupCount = sched.groups.size();
+
+    // Phase 1 — apply each group in isolation. Results are written into a
+    // pre-sized, index-keyed slot vector so the subsequent merge order is the
+    // (deterministic) schedule group order regardless of completion order.
+    std::vector<std::optional<std::pair<OpenView, std::size_t>>> slots(sched.groups.size());
+
+    unsigned const nThreads =
+        std::min<unsigned>(std::max<unsigned>(workers, 1u), static_cast<unsigned>(slots.size()));
+
+    if (nThreads <= 1)
+    {
+        for (std::size_t i = 0; i < sched.groups.size(); ++i)
+            slots[i].emplace(applyGroupIsolated(registry, closed, sched.groups[i].txns, j));
+    }
+    else
+    {
+        std::atomic<std::size_t> next{0};
+        auto worker = [&]() {
+            for (std::size_t i = next.fetch_add(1); i < sched.groups.size();
+                 i = next.fetch_add(1))
+                slots[i].emplace(applyGroupIsolated(registry, closed, sched.groups[i].txns, j));
+        };
+        std::vector<std::thread> pool;
+        pool.reserve(nThreads);
+        for (unsigned t = 0; t < nThreads; ++t)
+            pool.emplace_back(worker);
+        for (auto& th : pool)
+            th.join();
+    }
+
+    // Phase 2 — merge the disjoint write-sets into the target, in fixed group
+    // order. Sequential by design: `to` is a single mutable ledger.
+    for (auto& slot : slots)
+    {
+        slot->first.apply(to);
+        res.applied += slot->second;
+    }
+    return res;
+}
+
+}  // namespace xrpl

src/libxrpl/tx/Transactor.cpp

@@ -44,10 +44,12 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <cstdlib>
 #include <exception>
 #include <memory>
 #include <optional>
 #include <stdexcept>
+#include <string_view>
 #include <utility>
 #include <vector>
 
@@ -276,6 +278,21 @@ Transactor::Transactor(ApplyContext& ctx)
 {
 }
 
+AccessSet
+Transactor::commonAccountFootprint(STTx const& tx)
+{
+    AccessSet acc;
+    // The actor and the fee-payer (which differ only under delegation;
+    // getFeePayer() returns sfDelegate when present, else sfAccount).
+    acc.accounts.insert(keylet::account(tx.getAccountID(sfAccount)).key);
+    acc.accounts.insert(keylet::account(tx.getFeePayer()).key);
+    // A ticket-sequenced transaction consumes (erases) its Ticket object.
+    if (tx.isFieldPresent(sfTicketSequence))
+        acc.miscObjects.insert(
+            keylet::kTicket(tx.getAccountID(sfAccount), tx.getFieldU32(sfTicketSequence)).key);
+    return acc;
+}
+
 bool
 Transactor::validDataLength(std::optional<Slice> const& slice, std::size_t maxLength)
 {
@@ -284,6 +301,56 @@ Transactor::validDataLength(std::optional<Slice> const& slice, std::size_t maxLe
     return !slice->empty() && slice->length() <= maxLength;
 }
 
+#ifndef NDEBUG
+namespace {
+// Opt-in via the XRPL_ACCESS_AUDIT env var: log the measured footprint of
+// touchesGlobal transactors to feed the hard-transactor audit (Phase 1.5).
+bool
+accessSetAuditEnabled()
+{
+    static bool const enabled = [] {
+        auto const* v = std::getenv("XRPL_ACCESS_AUDIT");
+        return v != nullptr && std::string_view{v} != "" && std::string_view{v} != "0";
+    }();
+    return enabled;
+}
+}  // namespace
+
+void
+Transactor::verifyAccessSet() const
+{
+    // Qualify: the unqualified name would bind to the static member
+    // Transactor::accessSetOf (the global default), not the free dispatcher.
+    auto const declared = xrpl::accessSetOf(ctx_.tx, ctx_.baseView());
+
+    if (declared.touchesGlobal)
+    {
+        if (accessSetAuditEnabled())
+        {
+            std::size_t nonDir = 0;
+            for (auto const& [key, type] : ctx_.touchedEntries())
+                if (type != ltDIR_NODE)
+                    ++nonDir;
+            JLOG(j_.warn()) << "ACCESS_AUDIT txType=" << static_cast<int>(ctx_.tx.getTxnType())
+                            << " touched=" << ctx_.touchedEntries().size() << " nonDir=" << nonDir;
+        }
+        return;
+    }
+
+    auto const allowed = declared.keys();
+    for (auto const& [key, type] : ctx_.touchedEntries())
+    {
+        // Directory bookkeeping is out of scope for the access set (see
+        // AccessSet); a directory conflict is subsumed by its owner's account.
+        if (type == ltDIR_NODE)
+            continue;
+        XRPL_ASSERT(
+            allowed.contains(key),
+            "xrpl::Transactor::verifyAccessSet : touched entry within declared access set");
+    }
+}
+#endif
+
 std::uint32_t
 Transactor::getFlagsMask(PreflightContext const& ctx)
 {
@@ -1233,6 +1300,15 @@ Transactor::operator()()
     if (isTesSuccess(result))
         result = apply();
 
+#ifndef NDEBUG
+    // Validate (or audit) the declared access set against the actual footprint,
+    // but only on a clean success — tec/reset paths below intentionally touch
+    // extra state (removeUnfundedOffers, etc.) that the access set does not
+    // model.
+    if (isTesSuccess(result))
+        verifyAccessSet();
+#endif
+
     // No transaction can return temUNKNOWN from apply,
     // and it can't be passed in from a preclaim.
     XRPL_ASSERT(result != temUNKNOWN, "xrpl::Transactor::operator() : result is not temUNKNOWN");

src/libxrpl/tx/applySteps.cpp

@@ -420,6 +420,22 @@ calculateBaseFee(ReadView const& view, STTx const& tx)
     return invokeCalculateBaseFee(view, tx);
 }
 
+AccessSet
+accessSetOf(STTx const& tx, ReadView const& base)
+{
+    try
+    {
+        return withTxnType(base.rules(), tx.getTxnType(), [&]<typename T>() {
+            return T::accessSetOf(tx, base);
+        });
+    }
+    catch (UnknownTxnType const&)
+    {
+        // Unknown type — fail safe to global so a scheduler serializes it.
+        return AccessSet::global();
+    }
+}
+
 XRPAmount
 calculateDefaultBaseFee(ReadView const& view, STTx const& tx)
 {

src/libxrpl/tx/transactors/account/AccountSet.cpp

@@ -276,6 +276,13 @@ AccountSet::preclaim(PreclaimContext const& ctx)
     return tesSUCCESS;
 }
 
+AccessSet
+AccountSet::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // AccountSet only mutates the actor's own AccountRoot.
+    return commonAccountFootprint(tx);
+}
+
 TER
 AccountSet::doApply()
 {

src/libxrpl/tx/transactors/account/SetRegularKey.cpp

@@ -52,6 +52,13 @@ SetRegularKey::preflight(PreflightContext const& ctx)
     return tesSUCCESS;
 }
 
+AccessSet
+SetRegularKey::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // SetRegularKey only mutates the actor's own AccountRoot.
+    return commonAccountFootprint(tx);
+}
+
 TER
 SetRegularKey::doApply()
 {

src/libxrpl/tx/transactors/account/SignerListSet.cpp

@@ -107,6 +107,15 @@ SignerListSet::preflight(PreflightContext const& ctx)
     return tesSUCCESS;
 }
 
+AccessSet
+SignerListSet::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // Touches the actor's AccountRoot and its SignerList object.
+    AccessSet acc = commonAccountFootprint(tx);
+    acc.miscObjects.insert(keylet::signers(tx.getAccountID(sfAccount)).key);
+    return acc;
+}
+
 TER
 SignerListSet::doApply()
 {

src/libxrpl/tx/transactors/delegate/DelegateSet.cpp

@@ -65,6 +65,17 @@ DelegateSet::preclaim(PreclaimContext const& ctx)
     return tesSUCCESS;
 }
 
+AccessSet
+DelegateSet::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // Touches the actor's AccountRoot and the single Delegate object it owns
+    // for the authorized account.
+    AccessSet acc = commonAccountFootprint(tx);
+    acc.miscObjects.insert(
+        keylet::delegate(tx.getAccountID(sfAccount), tx[sfAuthorize]).key);
+    return acc;
+}
+
 TER
 DelegateSet::doApply()
 {

src/libxrpl/tx/transactors/did/DIDDelete.cpp

@@ -64,6 +64,15 @@ DIDDelete::deleteSLE(
     return tesSUCCESS;
 }
 
+AccessSet
+DIDDelete::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // Touches the actor's AccountRoot and its single DID object.
+    AccessSet acc = commonAccountFootprint(tx);
+    acc.miscObjects.insert(keylet::did(tx.getAccountID(sfAccount)).key);
+    return acc;
+}
+
 TER
 DIDDelete::doApply()
 {

src/libxrpl/tx/transactors/did/DIDSet.cpp

@@ -95,6 +95,15 @@ addSLE(ApplyContext& ctx, std::shared_ptr<SLE> const& sle, AccountID const& owne
     return tesSUCCESS;
 }
 
+AccessSet
+DIDSet::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // Touches the actor's AccountRoot and its single DID object.
+    AccessSet acc = commonAccountFootprint(tx);
+    acc.miscObjects.insert(keylet::did(tx.getAccountID(sfAccount)).key);
+    return acc;
+}
+
 TER
 DIDSet::doApply()
 {

src/libxrpl/tx/transactors/oracle/OracleDelete.cpp

@@ -79,6 +79,16 @@ OracleDelete::deleteOracle(
     return tesSUCCESS;
 }
 
+AccessSet
+OracleDelete::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // Touches the actor's AccountRoot and its own Oracle object.
+    AccessSet acc = commonAccountFootprint(tx);
+    acc.miscObjects.insert(
+        keylet::oracle(tx.getAccountID(sfAccount), tx[sfOracleDocumentID]).key);
+    return acc;
+}
+
 TER
 OracleDelete::doApply()
 {

src/libxrpl/tx/transactors/oracle/OracleSet.cpp

@@ -198,6 +198,16 @@ setPriceDataInnerObjTemplate(STObject& obj)
         obj.set(*elements);
 }
 
+AccessSet
+OracleSet::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // Touches the actor's AccountRoot and its own Oracle object.
+    AccessSet acc = commonAccountFootprint(tx);
+    acc.miscObjects.insert(
+        keylet::oracle(tx.getAccountID(sfAccount), tx[sfOracleDocumentID]).key);
+    return acc;
+}
+
 TER
 OracleSet::doApply()
 {

src/libxrpl/tx/transactors/payment/DepositPreauth.cpp

@@ -148,6 +148,31 @@ DepositPreauth::preclaim(PreclaimContext const& ctx)
     return tesSUCCESS;
 }
 
+AccessSet
+DepositPreauth::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // Touches the actor's AccountRoot and the single DepositPreauth object it
+    // creates or removes. The object key is fully derivable from the tx body
+    // (by authorized account, or by the sorted credential set).
+    AccessSet acc = commonAccountFootprint(tx);
+    auto const owner = tx.getAccountID(sfAccount);
+    if (tx.isFieldPresent(sfAuthorize))
+        acc.miscObjects.insert(keylet::depositPreauth(owner, tx[sfAuthorize]).key);
+    else if (tx.isFieldPresent(sfUnauthorize))
+        acc.miscObjects.insert(keylet::depositPreauth(owner, tx[sfUnauthorize]).key);
+    else if (tx.isFieldPresent(sfAuthorizeCredentials))
+        acc.miscObjects.insert(
+            keylet::depositPreauth(
+                owner, credentials::makeSorted(tx.getFieldArray(sfAuthorizeCredentials)))
+                .key);
+    else if (tx.isFieldPresent(sfUnauthorizeCredentials))
+        acc.miscObjects.insert(
+            keylet::depositPreauth(
+                owner, credentials::makeSorted(tx.getFieldArray(sfUnauthorizeCredentials)))
+                .key);
+    return acc;
+}
+
 TER
 DepositPreauth::doApply()
 {

src/libxrpl/tx/transactors/payment/Payment.cpp

@@ -395,6 +395,32 @@ Payment::preclaim(PreclaimContext const& ctx)
     return tesSUCCESS;
 }
 
+AccessSet
+Payment::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // Only the direct XRP->XRP case has a statically-enumerable footprint.
+    // Path / cross-currency / IOU / MPT payments route through the flow engine,
+    // which touches state-dependent trust lines and offers, so they stay global.
+    // A SendMax (even in XRP) also forces the rippling path.
+    if (tx.isFieldPresent(sfPaths) || tx.isFieldPresent(sfSendMax) ||
+        !tx.getFieldAmount(sfAmount).native())
+        return AccessSet::global();
+
+    AccessSet acc = commonAccountFootprint(tx);
+    auto const src = tx.getAccountID(sfAccount);
+    auto const dst = tx.getAccountID(sfDestination);
+    acc.accounts.insert(keylet::account(dst).key);
+    // The destination's deposit-authorization check may read this preauth
+    // object (declared unconditionally — pessimistic but a narrow conflict
+    // surface) and, when authorized by credentials, the credentials named in
+    // the tx.
+    acc.miscObjects.insert(keylet::depositPreauth(dst, src).key);
+    if (tx.isFieldPresent(sfCredentialIDs))
+        for (auto const& h : tx.getFieldV256(sfCredentialIDs))
+            acc.miscObjects.insert(h);
+    return acc;
+}
+
 TER
 Payment::doApply()
 {

src/libxrpl/tx/transactors/permissioned_domain/PermissionedDomainDelete.cpp

@@ -44,6 +44,16 @@ PermissionedDomainDelete::preclaim(PreclaimContext const& ctx)
     return tesSUCCESS;
 }
 
+AccessSet
+PermissionedDomainDelete::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // Touches the actor's AccountRoot and the single PermissionedDomain object
+    // named directly by the transaction.
+    AccessSet acc = commonAccountFootprint(tx);
+    acc.miscObjects.insert(keylet::permissionedDomain(tx[sfDomainID]).key);
+    return acc;
+}
+
 /** Attempt to delete the Permissioned Domain. */
 TER
 PermissionedDomainDelete::doApply()

src/libxrpl/tx/transactors/system/TicketCreate.cpp

@@ -64,6 +64,26 @@ TicketCreate::preclaim(PreclaimContext const& ctx)
     return tesSUCCESS;
 }
 
+AccessSet
+TicketCreate::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    AccessSet acc = commonAccountFootprint(tx);
+    auto const account = tx.getAccountID(sfAccount);
+    // Tickets are numbered from the account's current Sequence, which is ledger
+    // state — read it from the closed-ledger snapshot. The apply machinery
+    // advances Sequence by one before creating tickets for a sequence-based tx,
+    // so declare the inclusive range [seq, seq + count] to cover both the
+    // sequence-based (start = seq + 1) and ticket-based (start = seq) cases.
+    auto const sleAccount = base.read(keylet::account(account));
+    if (!sleAccount)
+        return AccessSet::global();
+    std::uint32_t const firstSeq = (*sleAccount)[sfSequence];
+    std::uint32_t const count = tx[sfTicketCount];
+    for (std::uint32_t i = 0; i <= count; ++i)
+        acc.miscObjects.insert(keylet::kTicket(account, firstSeq + i).key);
+    return acc;
+}
+
 TER
 TicketCreate::doApply()
 {

src/libxrpl/tx/transactors/token/TrustSet.cpp

@@ -323,6 +323,20 @@ TrustSet::preclaim(PreclaimContext const& ctx)
     return tesSUCCESS;
 }
 
+AccessSet
+TrustSet::accessSetOf(STTx const& tx, ReadView const& base)
+{
+    // Touches the actor and issuer AccountRoots and the single shared trust
+    // line between them; both endpoints come from sfLimitAmount.
+    AccessSet acc = commonAccountFootprint(tx);
+    auto const account = tx.getAccountID(sfAccount);
+    auto const limit = tx.getFieldAmount(sfLimitAmount);
+    auto const issuer = limit.getIssuer();
+    acc.accounts.insert(keylet::account(issuer).key);
+    acc.trustlines.insert(keylet::line(account, issuer, limit.get<Issue>().currency).key);
+    return acc;
+}
+
 TER
 TrustSet::doApply()
 {

src/tests/libxrpl/helpers/TxTest.h

@@ -347,6 +347,19 @@ public:
         return registry_;
     }
 
+    /**
+     * @brief Get the closed (base) ledger as a shared Ledger pointer.
+     *
+     * Exposes the underlying Ledger (not just the ReadView) so tests can build
+     * sibling ledgers from it — e.g. to apply a transaction set in a controlled
+     * order outside the canonicalizing close() path.
+     */
+    [[nodiscard]] std::shared_ptr<Ledger const>
+    getClosedLedgerPtr() const
+    {
+        return closedLedger_;
+    }
+
 private:
     TestServiceRegistry registry_;
     std::unordered_set<uint256, beast::Uhash<>> featureSet_;

src/tests/libxrpl/tx/AccessSet.cpp

@@ -0,0 +1,417 @@
+// Tests for per-transactor static access-set extraction (Plan 1, Phase 1).
+//
+// Two layers:
+//   1. AccessSet semantics (conflictsWith / keys) — pure, no ledger.
+//   2. accessSetOf(tx, view) content — the declared footprint of each migrated
+//      transactor matches expectation, and un-migrated / dynamic ones report
+//      touchesGlobal.
+//
+// The *subset* safety net (declared footprint ⊇ what apply actually touched) is
+// enforced continuously: in DEBUG builds Transactor::operator() asserts it on
+// every successful apply, so every env.submit() below — and every other test in
+// the suite that exercises a migrated transactor — validates it for free.
+
+#include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/Issue.h>
+#include <xrpl/protocol/SField.h>
+#include <xrpl/protocol/STArray.h>
+#include <xrpl/protocol/STObject.h>
+#include <xrpl/protocol/STTx.h>
+#include <xrpl/protocol/XRPAmount.h>
+#include <xrpl/protocol_autogen/transactions/AccountSet.h>
+#include <xrpl/protocol_autogen/transactions/DepositPreauth.h>
+#include <xrpl/protocol_autogen/transactions/OfferCreate.h>
+#include <xrpl/protocol_autogen/transactions/Payment.h>
+#include <xrpl/protocol_autogen/transactions/SetRegularKey.h>
+#include <xrpl/protocol_autogen/transactions/SignerListSet.h>
+#include <xrpl/protocol_autogen/transactions/TicketCreate.h>
+#include <xrpl/protocol_autogen/transactions/TrustSet.h>
+#include <xrpl/tx/AccessSet.h>
+#include <xrpl/tx/applySteps.h>
+
+#include <gtest/gtest.h>
+#include <helpers/Account.h>
+#include <helpers/IOU.h>
+#include <helpers/TxTest.h>
+
+#include <cstdint>
+#include <set>
+
+namespace xrpl::test {
+
+namespace {
+
+// Build a signed STTx from a builder with an explicit sequence (the signature
+// is irrelevant to accessSetOf; it never checks it).
+template <class Builder>
+std::shared_ptr<STTx const>
+sttxOf(Builder builder, Account const& signer, std::uint32_t seq)
+{
+    return builder.setSequence(seq).setFee(XRPAmount{10}).build(signer.pk(), signer.sk()).getSTTx();
+}
+
+std::set<uint256>
+keysOf(std::initializer_list<Keylet> ks)
+{
+    std::set<uint256> out;
+    for (auto const& k : ks)
+        out.insert(k.key);
+    return out;
+}
+
+}  // namespace
+
+//------------------------------------------------------------------------------
+// 1. AccessSet semantics
+//------------------------------------------------------------------------------
+
+TEST(AccessSet, ConflictDisjoint)
+{
+    Account const a("a");
+    Account const b("b");
+
+    AccessSet s1;
+    s1.accounts.insert(keylet::account(a.id()).key);
+    AccessSet s2;
+    s2.accounts.insert(keylet::account(b.id()).key);
+
+    EXPECT_FALSE(s1.conflictsWith(s2));
+    EXPECT_FALSE(s2.conflictsWith(s1));
+}
+
+TEST(AccessSet, ConflictSharedAccount)
+{
+    Account const a("a");
+    Account const b("b");
+
+    AccessSet s1;
+    s1.accounts.insert(keylet::account(a.id()).key);
+    s1.accounts.insert(keylet::account(b.id()).key);
+
+    AccessSet s2;  // shares account a
+    s2.accounts.insert(keylet::account(a.id()).key);
+
+    EXPECT_TRUE(s1.conflictsWith(s2));
+    EXPECT_TRUE(s2.conflictsWith(s1));
+}
+
+TEST(AccessSet, ConflictAcrossCategories)
+{
+    // The same key appearing under different categories still conflicts:
+    // conflict is decided on the flat union, not per-category.
+    Account const a("a");
+
+    AccessSet s1;
+    s1.accounts.insert(keylet::account(a.id()).key);
+    AccessSet s2;
+    s2.miscObjects.insert(keylet::account(a.id()).key);
+
+    EXPECT_TRUE(s1.conflictsWith(s2));
+}
+
+TEST(AccessSet, GlobalConflictsWithEverything)
+{
+    Account const a("a");
+    AccessSet local;
+    local.accounts.insert(keylet::account(a.id()).key);
+
+    AccessSet const g = AccessSet::global();
+    EXPECT_TRUE(g.touchesGlobal);
+    EXPECT_TRUE(g.conflictsWith(local));
+    EXPECT_TRUE(local.conflictsWith(g));
+    EXPECT_TRUE(g.conflictsWith(AccessSet{}));  // even with an empty set
+}
+
+TEST(AccessSet, KeysIsUnionAcrossCategories)
+{
+    Account const a("a");
+    Account const b("b");
+
+    AccessSet s;
+    s.accounts.insert(keylet::account(a.id()).key);
+    s.trustlines.insert(keylet::account(b.id()).key);  // any key; category is cosmetic
+    s.accounts.insert(keylet::account(a.id()).key);     // duplicate
+
+    EXPECT_EQ(s.keys().size(), 2u);
+}
+
+//------------------------------------------------------------------------------
+// 2. accessSetOf content
+//------------------------------------------------------------------------------
+
+TEST(AccessSet, AccountSetTouchesOnlyActor)
+{
+    TxTest env;
+    Account const alice("alice");
+    env.createAccount(alice, XRP(10000));
+    env.close();
+
+    auto const stx = sttxOf(
+        transactions::AccountSetBuilder{alice}, alice, env.getAccountRoot(alice.id()).getSequence());
+    auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger());
+
+    EXPECT_FALSE(acc.touchesGlobal);
+    EXPECT_EQ(acc.keys(), keysOf({keylet::account(alice.id())}));
+}
+
+TEST(AccessSet, SetRegularKeyTouchesOnlyActor)
+{
+    TxTest env;
+    Account const alice("alice");
+    Account const reg("reg");
+    env.createAccount(alice, XRP(10000));
+    env.close();
+
+    auto const stx = sttxOf(
+        transactions::SetRegularKeyBuilder{alice}.setRegularKey(reg.id()),
+        alice,
+        env.getAccountRoot(alice.id()).getSequence());
+    auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger());
+
+    EXPECT_FALSE(acc.touchesGlobal);
+    EXPECT_EQ(acc.keys(), keysOf({keylet::account(alice.id())}));
+}
+
+TEST(AccessSet, SignerListSetTouchesActorAndSignerList)
+{
+    TxTest env;
+    Account const alice("alice");
+    Account const bob("bob");
+    env.createAccount(alice, XRP(10000));
+    env.close();
+
+    STArray signerEntries(1);
+    signerEntries.push_back(STObject::makeInnerObject(sfSignerEntry));
+    signerEntries.back()[sfAccount] = bob.id();
+    signerEntries.back()[sfSignerWeight] = std::uint16_t{1};
+
+    auto const stx = sttxOf(
+        transactions::SignerListSetBuilder{alice, 1}.setSignerEntries(signerEntries),
+        alice,
+        env.getAccountRoot(alice.id()).getSequence());
+    auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger());
+
+    EXPECT_FALSE(acc.touchesGlobal);
+    EXPECT_EQ(acc.keys(), keysOf({keylet::account(alice.id()), keylet::signers(alice.id())}));
+}
+
+TEST(AccessSet, DepositPreauthTouchesActorAndPreauthObject)
+{
+    TxTest env;
+    Account const alice("alice");
+    Account const bob("bob");
+    env.createAccount(alice, XRP(10000));
+    env.close();
+
+    auto const stx = sttxOf(
+        transactions::DepositPreauthBuilder{alice}.setAuthorize(bob.id()),
+        alice,
+        env.getAccountRoot(alice.id()).getSequence());
+    auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger());
+
+    EXPECT_FALSE(acc.touchesGlobal);
+    EXPECT_EQ(
+        acc.keys(),
+        keysOf({keylet::account(alice.id()), keylet::depositPreauth(alice.id(), bob.id())}));
+}
+
+TEST(AccessSet, TrustSetTouchesBothEndpointsAndLine)
+{
+    TxTest env;
+    Account const alice("alice");
+    Account const gw("gateway");
+    IOU const usd("USD", gw);
+    env.createAccount(alice, XRP(10000));
+    env.createAccount(gw, XRP(10000));
+    env.close();
+
+    auto const stx = sttxOf(
+        transactions::TrustSetBuilder{alice}.setLimitAmount(usd.amount(10)),
+        alice,
+        env.getAccountRoot(alice.id()).getSequence());
+    auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger());
+
+    EXPECT_FALSE(acc.touchesGlobal);
+    EXPECT_EQ(
+        acc.keys(),
+        keysOf(
+            {keylet::account(alice.id()),
+             keylet::account(gw.id()),
+             keylet::line(alice.id(), gw.id(), usd.issue().currency)}));
+}
+
+TEST(AccessSet, PaymentXrpDirectTouchesSrcDstAndPreauth)
+{
+    TxTest env;
+    Account const alice("alice");
+    Account const bob("bob");
+    env.createAccount(alice, XRP(10000));
+    env.createAccount(bob, XRP(10000));
+    env.close();
+
+    auto const stx = sttxOf(
+        transactions::PaymentBuilder{alice, bob, XRP(1)},
+        alice,
+        env.getAccountRoot(alice.id()).getSequence());
+    auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger());
+
+    EXPECT_FALSE(acc.touchesGlobal);
+    EXPECT_EQ(
+        acc.keys(),
+        keysOf(
+            {keylet::account(alice.id()),
+             keylet::account(bob.id()),
+             keylet::depositPreauth(bob.id(), alice.id())}));
+}
+
+TEST(AccessSet, TicketCreateDeclaresSequenceRange)
+{
+    TxTest env;
+    Account const alice("alice");
+    env.createAccount(alice, XRP(10000));
+    env.close();
+
+    std::uint32_t const seq = env.getAccountRoot(alice.id()).getSequence();
+    std::uint32_t const count = 3;
+
+    auto const stx =
+        sttxOf(transactions::TicketCreateBuilder{alice, count}, alice, seq);
+    auto const acc = xrpl::accessSetOf(*stx, env.getClosedLedger());
+
+    EXPECT_FALSE(acc.touchesGlobal);
+    // Inclusive range [seq, seq + count] covers both sequence- and ticket-based
+    // apply (the machinery may advance the sequence by one before creating).
+    std::set<uint256> expected{keylet::account(alice.id()).key};
+    for (std::uint32_t i = 0; i <= count; ++i)
+        expected.insert(keylet::kTicket(alice.id(), seq + i).key);
+    EXPECT_EQ(acc.keys(), expected);
+}
+
+//------------------------------------------------------------------------------
+// 3. Fail-safe: dynamic-footprint transactions report touchesGlobal
+//------------------------------------------------------------------------------
+
+TEST(AccessSet, PaymentWithSendMaxIsGlobal)
+{
+    // Even an all-XRP payment with SendMax routes through the flow engine.
+    TxTest env;
+    Account const alice("alice");
+    Account const bob("bob");
+    env.createAccount(alice, XRP(10000));
+    env.createAccount(bob, XRP(10000));
+    env.close();
+
+    auto const stx = sttxOf(
+        transactions::PaymentBuilder{alice, bob, XRP(1)}.setSendMax(XRP(2)),
+        alice,
+        env.getAccountRoot(alice.id()).getSequence());
+    EXPECT_TRUE(xrpl::accessSetOf(*stx, env.getClosedLedger()).touchesGlobal);
+}
+
+TEST(AccessSet, IouPaymentIsGlobal)
+{
+    TxTest env;
+    Account const gw("gateway");
+    Account const alice("alice");
+    IOU const usd("USD", gw);
+    env.createAccount(gw, XRP(10000), asfDefaultRipple);
+    env.createAccount(alice, XRP(10000), asfDefaultRipple);
+    env.close();
+
+    auto const stx = sttxOf(
+        transactions::PaymentBuilder{gw, alice, usd.amount(5)},
+        gw,
+        env.getAccountRoot(gw.id()).getSequence());
+    EXPECT_TRUE(xrpl::accessSetOf(*stx, env.getClosedLedger()).touchesGlobal);
+}
+
+TEST(AccessSet, UnmigratedTransactorIsGlobal)
+{
+    // OfferCreate is intentionally not migrated (offer crossing has a
+    // state-dependent footprint); it must fall back to the global default.
+    TxTest env;
+    Account const alice("alice");
+    Account const gw("gateway");
+    IOU const usd("USD", gw);
+    env.createAccount(alice, XRP(10000));
+    env.close();
+
+    auto const stx = sttxOf(
+        transactions::OfferCreateBuilder{alice, usd.amount(10), XRP(10)},
+        alice,
+        env.getAccountRoot(alice.id()).getSequence());
+    EXPECT_TRUE(xrpl::accessSetOf(*stx, env.getClosedLedger()).touchesGlobal);
+}
+
+//------------------------------------------------------------------------------
+// 4. Subset safety net (explicit). In DEBUG these apply paths assert the
+//    declared footprint ⊇ the touched footprint; tesSUCCESS means it held.
+//------------------------------------------------------------------------------
+
+TEST(AccessSet, SubsetPaymentToNewAccount)
+{
+    // Exercises the absence-probe path: the brand-new destination is read
+    // (returns nullptr) and then inserted; both must fall within the declared
+    // {src, dst, depositPreauth(dst, src)}.
+    TxTest env;
+    Account const alice("alice");
+    Account const carol("carol");  // never created
+    env.createAccount(alice, XRP(10000));
+    env.close();
+
+    EXPECT_EQ(
+        env.submit(transactions::PaymentBuilder{alice, carol, XRP(100)}, alice).ter, tesSUCCESS);
+}
+
+TEST(AccessSet, SubsetTicketCreateThenUse)
+{
+    TxTest env;
+    Account const alice("alice");
+    env.createAccount(alice, XRP(10000));
+    env.close();
+
+    std::uint32_t const seq = env.getAccountRoot(alice.id()).getSequence();
+    EXPECT_EQ(env.submit(transactions::TicketCreateBuilder{alice, 1}, alice).ter, tesSUCCESS);
+    env.close();
+
+    // Use the ticket (a ticket-based AccountSet) — exercises common-footprint
+    // ticket handling on the consume side.
+    std::uint32_t const ticketSeq = seq + 1;
+    EXPECT_EQ(
+        env.submit(transactions::AccountSetBuilder{alice}.setTicketSequence(ticketSeq), alice).ter,
+        tesSUCCESS);
+}
+
+TEST(AccessSet, SubsetTrustSetAndSignerList)
+{
+    TxTest env;
+    Account const alice("alice");
+    Account const bob("bob");
+    Account const gw("gateway");
+    IOU const usd("USD", gw);
+    env.createAccount(alice, XRP(10000), asfDefaultRipple);
+    env.createAccount(gw, XRP(10000), asfDefaultRipple);
+    env.close();
+
+    EXPECT_EQ(
+        env.submit(transactions::TrustSetBuilder{alice}.setLimitAmount(usd.amount(10)), alice).ter,
+        tesSUCCESS);
+    env.close();
+
+    STArray signerEntries(1);
+    signerEntries.push_back(STObject::makeInnerObject(sfSignerEntry));
+    signerEntries.back()[sfAccount] = bob.id();
+    signerEntries.back()[sfSignerWeight] = std::uint16_t{1};
+
+    EXPECT_EQ(
+        env.submit(
+               transactions::SignerListSetBuilder{alice, 1}.setSignerEntries(signerEntries), alice)
+            .ter,
+        tesSUCCESS);
+    env.close();
+
+    // Remove the signer list (destroy path).
+    EXPECT_EQ(env.submit(transactions::SignerListSetBuilder{alice, 0}, alice).ter, tesSUCCESS);
+}
+
+}  // namespace xrpl::test

src/tests/libxrpl/tx/Schedule.cpp

@@ -0,0 +1,190 @@
+// Tests for scheduleApply (Plan 1, Phase 2): partitioning a canonical
+// transaction set into independent conflict groups for parallel application.
+//
+// The headline guarantee is the differential test: applying a workload in the
+// scheduler's (reordered) group order produces a byte-identical account-state
+// root to applying it in canonical order. That is the correctness contract a
+// parallel executor depends on.
+
+#include <xrpl/protocol/SField.h>
+#include <xrpl/protocol/STTx.h>
+#include <xrpl/protocol/XRPAmount.h>
+#include <xrpl/protocol_autogen/transactions/AccountSet.h>
+#include <xrpl/protocol_autogen/transactions/OfferCreate.h>
+#include <xrpl/protocol_autogen/transactions/Payment.h>
+#include <xrpl/tx/Schedule.h>
+#include <xrpl/tx/applySteps.h>
+
+#include <gtest/gtest.h>
+#include <helpers/Account.h>
+#include <helpers/IOU.h>
+#include <helpers/TxTest.h>
+
+#include <cstdint>
+#include <memory>
+#include <vector>
+
+namespace xrpl::test {
+
+namespace {
+
+// Build a signed Payment STTx with an explicit sequence.
+std::shared_ptr<STTx const>
+payment(Account const& from, Account const& to, STAmount const& amount, std::uint32_t seq)
+{
+    return transactions::PaymentBuilder{from, to, amount}
+        .setSequence(seq)
+        .setFee(XRPAmount{10})
+        .build(from.pk(), from.sk())
+        .getSTTx();
+}
+
+}  // namespace
+
+TEST(Schedule, DisjointPaymentsFormSeparateGroups)
+{
+    TxTest env;
+    Account const a("a"), b("b"), c("c"), d("d"), e("e"), f("f");
+    for (auto const* acct : {&a, &b, &c, &d, &e, &f})
+        env.createAccount(*acct, XRP(10000));
+    env.close();
+
+    std::vector<std::shared_ptr<STTx const>> txns{
+        payment(a, b, XRP(1), env.getAccountRoot(a.id()).getSequence()),
+        payment(c, d, XRP(1), env.getAccountRoot(c.id()).getSequence()),
+        payment(e, f, XRP(1), env.getAccountRoot(e.id()).getSequence()),
+    };
+
+    auto const sched = scheduleApply(txns, env.getClosedLedger());
+
+    EXPECT_FALSE(sched.fullySerial);
+    EXPECT_EQ(sched.groups.size(), 3u);
+    EXPECT_EQ(sched.size(), 3u);
+    for (auto const& g : sched.groups)
+        EXPECT_EQ(g.txns.size(), 1u);
+}
+
+TEST(Schedule, SameSourceFormsOneOrderedGroup)
+{
+    TxTest env;
+    Account const a("a"), b("b"), c("c");
+    for (auto const* acct : {&a, &b, &c})
+        env.createAccount(*acct, XRP(10000));
+    env.close();
+
+    std::uint32_t const seq = env.getAccountRoot(a.id()).getSequence();
+    std::vector<std::shared_ptr<STTx const>> txns{
+        payment(a, b, XRP(1), seq),
+        payment(a, c, XRP(1), seq + 1),
+    };
+
+    auto const sched = scheduleApply(txns, env.getClosedLedger());
+
+    ASSERT_EQ(sched.groups.size(), 1u);
+    ASSERT_EQ(sched.groups[0].txns.size(), 2u);
+    // Canonical (sequence) order preserved within the group.
+    EXPECT_EQ(sched.groups[0].txns[0]->getSeqValue(), seq);
+    EXPECT_EQ(sched.groups[0].txns[1]->getSeqValue(), seq + 1);
+}
+
+TEST(Schedule, SharedDestinationConflicts)
+{
+    TxTest env;
+    Account const a("a"), b("b"), z("z");
+    for (auto const* acct : {&a, &b, &z})
+        env.createAccount(*acct, XRP(10000));
+    env.close();
+
+    // Both pay the same destination z -> they share z's AccountRoot -> 1 group.
+    std::vector<std::shared_ptr<STTx const>> txns{
+        payment(a, z, XRP(1), env.getAccountRoot(a.id()).getSequence()),
+        payment(b, z, XRP(1), env.getAccountRoot(b.id()).getSequence()),
+    };
+
+    auto const sched = scheduleApply(txns, env.getClosedLedger());
+    EXPECT_FALSE(sched.fullySerial);
+    ASSERT_EQ(sched.groups.size(), 1u);
+    EXPECT_EQ(sched.groups[0].txns.size(), 2u);
+}
+
+TEST(Schedule, GlobalTransactionForcesFullySerial)
+{
+    TxTest env;
+    Account const a("a"), b("b"), gw("gw");
+    IOU const usd("USD", gw);
+    for (auto const* acct : {&a, &b, &gw})
+        env.createAccount(*acct, XRP(10000));
+    env.close();
+
+    // An OfferCreate is touchesGlobal (dynamic footprint) -> whole set serial.
+    auto const offer = transactions::OfferCreateBuilder{a, usd.amount(10), XRP(10)}
+                           .setSequence(env.getAccountRoot(a.id()).getSequence())
+                           .setFee(XRPAmount{10})
+                           .build(a.pk(), a.sk())
+                           .getSTTx();
+
+    std::vector<std::shared_ptr<STTx const>> txns{
+        payment(b, gw, XRP(1), env.getAccountRoot(b.id()).getSequence()),
+        offer,
+    };
+
+    auto const sched = scheduleApply(txns, env.getClosedLedger());
+    EXPECT_TRUE(sched.fullySerial);
+    EXPECT_TRUE(sched.groups.empty());
+    EXPECT_EQ(sched.serial.size(), 2u);
+}
+
+// The headline correctness property the scheduler must guarantee: any two
+// transactions placed in DIFFERENT groups have non-conflicting access sets.
+// Together with the (separately, continuously verified) fact that each access
+// set is a superset of what the transaction actually touches, this is exactly
+// what makes applying distinct groups concurrently state-equivalent to a serial
+// apply — independent writes commute. We assert the partition property directly,
+// which is stronger and more honest than an apply-order replay through this test
+// harness (whose close() re-canonicalizes by tx hash regardless of submission
+// order, so it cannot observe a reordering). The full apply-in-schedule-order
+// state-root differential belongs to Phase 3, where a parallel executor applies
+// outside the canonicalizing close path.
+TEST(Schedule, GroupsArePairwiseIndependent)
+{
+    TxTest env;
+    Account const a("a"), b("b"), c("c"), d("d"), e("e"), f("f"), z("z");
+    for (auto const* acct : {&a, &b, &c, &d, &e, &f, &z})
+        env.createAccount(*acct, XRP(10000));
+    env.close();
+
+    std::uint32_t const seqA = env.getAccountRoot(a.id()).getSequence();
+    std::vector<std::shared_ptr<STTx const>> txns{
+        payment(a, b, XRP(7), seqA),
+        payment(c, d, XRP(3), env.getAccountRoot(c.id()).getSequence()),
+        payment(e, f, XRP(5), env.getAccountRoot(e.id()).getSequence()),
+        payment(a, b, XRP(2), seqA + 1),  // same source a -> same group as #1
+        payment(c, z, XRP(1), env.getAccountRoot(c.id()).getSequence() + 1),  // shares c -> #2
+    };
+
+    auto const& base = env.getClosedLedger();
+    auto const sched = scheduleApply(txns, base);
+    ASSERT_FALSE(sched.fullySerial);
+
+    // Every transaction is scheduled exactly once.
+    EXPECT_EQ(sched.size(), txns.size());
+
+    // Within a group, canonical (input) order is preserved — verify per-source
+    // sequences are monotonic.
+    for (auto const& g : sched.groups)
+        for (std::size_t i = 1; i < g.txns.size(); ++i)
+            if (g.txns[i]->getAccountID(sfAccount) == g.txns[i - 1]->getAccountID(sfAccount))
+                EXPECT_LT(g.txns[i - 1]->getSeqValue(), g.txns[i]->getSeqValue());
+
+    // The core invariant: any two transactions in DIFFERENT groups do not
+    // conflict. (Equivalently: every real conflict is contained within a group.)
+    for (std::size_t i = 0; i < sched.groups.size(); ++i)
+        for (std::size_t j = i + 1; j < sched.groups.size(); ++j)
+            for (auto const& ta : sched.groups[i].txns)
+                for (auto const& tb : sched.groups[j].txns)
+                    EXPECT_FALSE(
+                        accessSetOf(*ta, base).conflictsWith(accessSetOf(*tb, base)))
+                        << "transactions in different groups must not conflict";
+}
+
+}  // namespace xrpl::test

src/tests/libxrpl/tx/ScheduledApply.cpp

@@ -0,0 +1,254 @@
+// Differential test for applyScheduled (Plan 1, Phase 3 core): applying a
+// transaction set via its conflict-group schedule (each group isolated over the
+// closed snapshot, write-sets merged) must yield a byte-identical account-state
+// root to a serial canonical apply.
+//
+// Unlike a test routed through TxTest::close() (which re-canonicalizes by tx
+// hash and so cannot observe a reordering), this test builds BOTH ledgers
+// itself, so the comparison is real: if the scheduler ever placed two
+// conflicting transactions in different groups, the roots would diverge here.
+
+#include <xrpl/core/ServiceRegistry.h>
+#include <xrpl/ledger/Ledger.h>
+#include <xrpl/ledger/OpenView.h>
+#include <xrpl/protocol/STTx.h>
+#include <xrpl/protocol/XRPAmount.h>
+#include <xrpl/protocol_autogen/transactions/OfferCreate.h>
+#include <xrpl/protocol_autogen/transactions/Payment.h>
+#include <xrpl/tx/Schedule.h>
+#include <xrpl/tx/apply.h>
+
+#include <gtest/gtest.h>
+#include <helpers/Account.h>
+#include <helpers/IOU.h>
+#include <helpers/TxTest.h>
+
+#include <chrono>
+#include <cstdint>
+#include <cstdio>
+#include <memory>
+#include <string>
+#include <vector>
+
+namespace xrpl::test {
+
+namespace {
+
+std::shared_ptr<STTx const>
+payment(Account const& from, Account const& to, STAmount const& amount, std::uint32_t seq)
+{
+    return transactions::PaymentBuilder{from, to, amount}
+        .setSequence(seq)
+        .setFee(XRPAmount{10})
+        .build(from.pk(), from.sk())
+        .getSTTx();
+}
+
+// Build a fresh ledger from `closed`, apply `txns` in the given exact order in a
+// single accumulating view (the serial ground truth), and return its
+// account-state root.
+uint256
+serialStateRoot(
+    TxTest& env,
+    Ledger const& closed,
+    std::vector<std::shared_ptr<STTx const>> const& txns)
+{
+    auto const closeTime = env.getCloseTime() + closed.header().closeTimeResolution;
+    auto next = std::make_shared<Ledger>(closed, closeTime);
+    {
+        OpenView accum(&closed);
+        for (auto const& tx : txns)
+            apply(env.getServiceRegistry(), accum, *tx, TapNone, env.getServiceRegistry().getJournal("apply"));
+        accum.apply(*next);
+    }
+    next->setAccepted(closeTime, closed.header().closeTimeResolution, true);
+    return next->header().accountHash;
+}
+
+// Build a fresh ledger from `closed`, apply `txns` via applyScheduled (grouped +
+// merged), and return its account-state root, along with the schedule result.
+std::pair<uint256, ScheduledApplyResult>
+scheduledStateRoot(
+    TxTest& env,
+    Ledger const& closed,
+    std::vector<std::shared_ptr<STTx const>> const& txns,
+    unsigned workers = 1)
+{
+    auto const closeTime = env.getCloseTime() + closed.header().closeTimeResolution;
+    auto next = std::make_shared<Ledger>(closed, closeTime);
+    auto const res = applyScheduled(
+        env.getServiceRegistry(),
+        closed,
+        *next,
+        txns,
+        env.getServiceRegistry().getJournal("apply"),
+        workers);
+    next->setAccepted(closeTime, closed.header().closeTimeResolution, true);
+    return {next->header().accountHash, res};
+}
+
+}  // namespace
+
+TEST(ScheduledApply, ParallelGroupedMatchesSerial)
+{
+    TxTest env;
+    Account const a("a"), b("b"), c("c"), d("d"), e("e"), f("f"), g("g");
+    for (auto const* acct : {&a, &b, &c, &d, &e, &f, &g})
+        env.createAccount(*acct, XRP(10000));
+    env.close();
+
+    auto const& closed = *env.getClosedLedgerPtr();
+    std::uint32_t const seqA = env.getAccountRoot(a.id()).getSequence();
+
+    // Groups: {p1,p4} (source a, ordered), {p2}, {p3} — three independent groups.
+    std::vector<std::shared_ptr<STTx const>> txns{
+        payment(a, b, XRP(7), seqA),
+        payment(c, d, XRP(3), env.getAccountRoot(c.id()).getSequence()),
+        payment(e, f, XRP(5), env.getAccountRoot(e.id()).getSequence()),
+        payment(a, g, XRP(2), seqA + 1),
+    };
+
+    auto const serialRoot = serialStateRoot(env, closed, txns);
+    auto const [scheduledRoot, res] = scheduledStateRoot(env, closed, txns);
+
+    EXPECT_FALSE(res.fullySerial);
+    EXPECT_EQ(res.groupCount, 3u);
+    EXPECT_EQ(res.applied, 4u);
+
+    // The state root must be non-trivial (real accounts changed) AND identical
+    // regardless of the parallel grouping — the determinism guarantee.
+    EXPECT_NE(serialRoot, uint256{});
+    EXPECT_EQ(serialRoot, scheduledRoot);
+}
+
+TEST(ScheduledApply, ThreadedMatchesSerialAcrossManyGroups)
+{
+    TxTest env;
+    // 24 accounts -> 12 disjoint payment pairs -> 12 independent groups, applied
+    // across a thread pool. Repeated to give nondeterminism/races a chance to
+    // surface. (A clean unit pass is necessary, not sufficient, for production —
+    // see applyScheduled's note on certification.)
+    constexpr int kPairs = 12;
+    std::vector<Account> accts;
+    accts.reserve(kPairs * 2);
+    for (int i = 0; i < kPairs * 2; ++i)
+        accts.emplace_back("acct" + std::to_string(i));
+    for (auto const& a : accts)
+        env.createAccount(a, XRP(10000));
+    env.close();
+
+    auto const& closed = *env.getClosedLedgerPtr();
+    std::vector<std::shared_ptr<STTx const>> txns;
+    for (int p = 0; p < kPairs; ++p)
+    {
+        auto const& from = accts[2 * p];
+        auto const& to = accts[2 * p + 1];
+        txns.push_back(
+            payment(from, to, XRP(p + 1), env.getAccountRoot(from.id()).getSequence()));
+    }
+
+    auto const serialRoot = serialStateRoot(env, closed, txns);
+    EXPECT_NE(serialRoot, uint256{});
+
+    for (int iter = 0; iter < 8; ++iter)
+    {
+        auto const [threadedRoot, res] = scheduledStateRoot(env, closed, txns, /*workers=*/8);
+        EXPECT_FALSE(res.fullySerial);
+        EXPECT_EQ(res.groupCount, static_cast<std::size_t>(kPairs));
+        EXPECT_EQ(res.applied, static_cast<std::size_t>(kPairs));
+        EXPECT_EQ(threadedRoot, serialRoot) << "threaded apply diverged on iteration " << iter;
+    }
+}
+
+TEST(ScheduledApply, FullySerialPathAlsoMatches)
+{
+    TxTest env;
+    Account const a("a"), b("b"), gw("gw");
+    IOU const usd("USD", gw);
+    for (auto const* acct : {&a, &b, &gw})
+        env.createAccount(*acct, XRP(10000));
+    env.close();
+
+    auto const& closed = *env.getClosedLedgerPtr();
+
+    // An OfferCreate is touchesGlobal -> scheduleApply falls back to fully
+    // serial; applyScheduled then applies the whole set in canonical order.
+    auto const offer = transactions::OfferCreateBuilder{a, usd.amount(10), XRP(10)}
+                           .setSequence(env.getAccountRoot(a.id()).getSequence())
+                           .setFee(XRPAmount{10})
+                           .build(a.pk(), a.sk())
+                           .getSTTx();
+    std::vector<std::shared_ptr<STTx const>> txns{
+        payment(b, gw, XRP(1), env.getAccountRoot(b.id()).getSequence()),
+        offer,
+    };
+
+    auto const serialRoot = serialStateRoot(env, closed, txns);
+    auto const [scheduledRoot, res] = scheduledStateRoot(env, closed, txns);
+
+    EXPECT_TRUE(res.fullySerial);
+    EXPECT_NE(serialRoot, uint256{});
+    EXPECT_EQ(serialRoot, scheduledRoot);
+}
+
+// Throughput benchmark: time applyScheduled over many disjoint payments at
+// increasing worker counts. Run explicitly:
+//   xrpl.test.tx --gtest_filter='ScheduledApply.ThroughputBenchmark'
+// Build type matters enormously — Debug numbers (assertions on, unoptimized) are
+// directional only; use a Release build for representative figures.
+TEST(ScheduledApply, ThroughputBenchmark)
+{
+    constexpr int kPairs = 400;   // -> kPairs independent groups, 2*kPairs accounts
+    constexpr int kReps = 5;
+
+    TxTest env;
+    std::vector<Account> accts;
+    accts.reserve(kPairs * 2);
+    for (int i = 0; i < kPairs * 2; ++i)
+        accts.emplace_back("ba" + std::to_string(i));
+    // Fund in batches with a single close per batch (createAccount closes each).
+    for (auto const& a : accts)
+        env.createAccount(a, XRP(10000));
+    env.close();
+
+    auto const& closed = *env.getClosedLedgerPtr();
+    std::vector<std::shared_ptr<STTx const>> txns;
+    txns.reserve(kPairs);
+    for (int p = 0; p < kPairs; ++p)
+        txns.push_back(payment(
+            accts[2 * p], accts[2 * p + 1], XRP(1), env.getAccountRoot(accts[2 * p].id()).getSequence()));
+
+    auto const closeTime = env.getCloseTime() + closed.header().closeTimeResolution;
+    auto bestNanos = [&](unsigned workers) {
+        long long best = -1;
+        for (int r = 0; r < kReps; ++r)
+        {
+            auto next = std::make_shared<Ledger>(closed, closeTime);
+            auto const t0 = std::chrono::steady_clock::now();
+            auto const res = applyScheduled(
+                env.getServiceRegistry(), closed, *next, txns,
+                env.getServiceRegistry().getJournal("bench"), workers);
+            auto const t1 = std::chrono::steady_clock::now();
+            EXPECT_EQ(res.applied, static_cast<std::size_t>(kPairs));
+            auto const ns = std::chrono::duration_cast<std::chrono::nanoseconds>(t1 - t0).count();
+            if (best < 0 || ns < best)
+                best = ns;
+        }
+        return best;
+    };
+
+    std::printf("\n=== applyScheduled throughput: %d disjoint payments (best of %d) ===\n",
+                kPairs, kReps);
+    long long base = 0;
+    for (unsigned w : {1u, 2u, 4u, 8u})
+    {
+        auto const ns = bestNanos(w);
+        if (w == 1)
+            base = ns;
+        std::printf("  workers=%u: %8.3f ms total, %7.1f us/tx, speedup %.2fx\n",
+                    w, ns / 1e6, ns / 1e3 / kPairs, base / double(ns));
+    }
+    std::printf("(build type dominates these numbers; Debug is directional only)\n");
+}
+
+}  // namespace xrpl::test

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

@@ -17,12 +17,18 @@
 #include <xrpl/protocol/LedgerHeader.h>
 #include <xrpl/protocol/Protocol.h>
 #include <xrpl/protocol/SystemParameters.h>
+#include <xrpl/tx/Schedule.h>
 #include <xrpl/tx/apply.h>
 
+#include <algorithm>
 #include <cstddef>
+#include <cstdlib>
 #include <exception>
 #include <memory>
 #include <set>
+#include <string_view>
+#include <thread>
+#include <vector>
 
 namespace xrpl {
 
@@ -90,6 +96,39 @@ buildLedgerImpl(
   @return number of transactions applied; transactions to retry left in txns
 */
 
+namespace {
+
+// Experimental, operator-opt-in parallel apply (Plan 1). Gated by the
+// XRPL_PARALLEL_APPLY env var; default OFF, so default behaviour is unchanged.
+// A proper amendment/Config flag is the productionization step.
+bool
+parallelApplyEnabled()
+{
+    static bool const enabled = [] {
+        auto const* v = std::getenv("XRPL_PARALLEL_APPLY");
+        return v != nullptr && std::string_view{v} != "" && std::string_view{v} != "0";
+    }();
+    return enabled;
+}
+
+unsigned
+parallelApplyWorkers()
+{
+    static unsigned const workers = [] {
+        if (auto const* v = std::getenv("XRPL_PARALLEL_APPLY_WORKERS"))
+        {
+            auto const n = std::atoi(v);
+            if (n > 0)
+                return static_cast<unsigned>(n);
+        }
+        unsigned const hw = std::thread::hardware_concurrency();
+        return std::clamp<unsigned>(hw > 2 ? hw - 2 : 2, 2u, 8u);
+    }();
+    return workers;
+}
+
+}  // namespace
+
 std::size_t
 applyTransactions(
     Application& app,
@@ -99,6 +138,36 @@ applyTransactions(
     OpenView& view,
     beast::Journal j)
 {
+    // Plan 1 parallel apply: schedule the canonical set into independent groups
+    // and apply them (optionally across a thread pool), merging the disjoint
+    // write-sets into `view`. Behind XRPL_PARALLEL_APPLY; the resulting state is
+    // byte-identical to the serial path for conflict-free reorderings (the
+    // schedule guarantees this; see ScheduledApply differential tests). On flag
+    // ledgers / any global tx, the scheduler falls back to fully serial.
+    if (parallelApplyEnabled())
+    {
+        std::vector<std::shared_ptr<STTx const>> ordered;
+        ordered.reserve(txns.size());
+        for (auto const& item : txns)
+        {
+            if (built->txExists(item.first.getTXID()))
+                continue;
+            ordered.push_back(item.second);
+        }
+
+        auto const res = applyScheduled(app, *built, view, ordered, j, parallelApplyWorkers());
+
+        JLOG(j.debug()) << "Parallel apply: " << res.applied << " applied across "
+                        << res.groupCount << " group(s)"
+                        << (res.fullySerial ? " (fell back to serial)" : "");
+
+        // Every transaction has been consumed; the parallel path needs no retry
+        // passes — by access-set disjointness, no cross-group dependency exists.
+        for (auto it = txns.begin(); it != txns.end();)
+            it = txns.erase(it);
+        return res.applied;
+    }
+
     bool certainRetry = true;
     std::size_t count = 0;
 

tasks/plan-1-status.md

@@ -0,0 +1,200 @@
+# Plan 1 (parallel transaction apply) — implementation status
+
+Branch: `feat/parallel-apply-access-set`. This file tracks what is implemented,
+verified, and what remains, so any engineer can continue without re-deriving.
+
+## Implemented & verified
+
+### Phase 1 foundation (commit "static access-set extraction + DEBUG assertion gate")
+- `include/xrpl/tx/AccessSet.h` — the declared per-tx footprint (categorised key
+  sets + `touchesGlobal`), `keys()`, `conflictsWith()`.
+- `Transactor::accessSetOf(STTx, ReadView)` — name-hidden dispatch hook, base
+  default `touchesGlobal=true` (fail-safe). Free `accessSetOf` dispatcher in
+  `applySteps.cpp`. `Transactor::commonAccountFootprint` helper.
+- DEBUG touched-key instrumentation in `detail::ApplyStateTable` (the single
+  chokepoint for read/exists/peek/insert/update/erase), exposed up through
+  `ApplyViewBase`/`ApplyContext`.
+- DEBUG subset assertion + `XRPL_ACCESS_AUDIT` footprint dump in
+  `Transactor::operator()` (success path only).
+- Tests: `src/tests/libxrpl/tx/AccessSet.cpp` (18). Verified: full
+  `xrpl.test.protocol_autogen` (495 tests) passes with the assertion live.
+
+### Phase 2 scheduler (commit "access-set scheduler ... independent groups")
+- `include/xrpl/tx/Schedule.h` + `src/libxrpl/tx/Schedule.cpp` — `scheduleApply`
+  partitions a canonical tx set into independent `ConflictGroup`s via union-find
+  over an inverted key→tx index. Any `touchesGlobal` tx ⇒ conservative
+  fully-serial fallback (flag-ledger handling).
+- Tests: `src/tests/libxrpl/tx/Schedule.cpp` (5), incl. the core invariant
+  **GroupsArePairwiseIndependent** (txns in different groups never conflict).
+
+### Migrated transactors (13) — `accessSetOf` declared, assertion-verified
+AccountSet, SetRegularKey, DepositPreauth, SignerListSet, TicketCreate, TrustSet,
+DIDSet, DIDDelete, Payment (XRP→XRP only), OracleSet, OracleDelete, DelegateSet,
+PermissionedDomainDelete.
+
+## Load-bearing design rules (apply to every future migration)
+1. **Directory exclusion + owner declaration.** The assertion excludes
+   `ltDIR_NODE` entries (owner-dir pages are derived bookkeeping). This is sound
+   ONLY if, for every owner directory a transactor modifies, that owner's
+   `keylet::account` is declared. So: whenever `doApply` does `dirInsert`/
+   `dirRemove` on `ownerDir(X)`, declare `account(X)` — even if X's root isn't
+   otherwise written (e.g. CredentialCreate touches the subject's owner dir).
+2. **Shared (non-owner) directories ⇒ global.** Book directories and NFT
+   buy/sell directories (`nftSells`/`nftBuys`, keyed by NFTokenID) are shared
+   across accounts; they are genuine cross-account conflict surfaces with no
+   single owning account, so any transactor touching them stays `touchesGlobal`.
+3. **`succ()`/range scans ⇒ not statically declarable.** A footprint discovered
+   by walking a chain (NFT page chains) is not a static superset ⇒ global.
+4. **Snapshot reads are allowed.** `accessSetOf(tx, base)` may read `base` to
+   resolve a field stored inside an object SLE (e.g. an escrow's Destination).
+5. The DEBUG subset assertion is the gate: migrate, run `xrpl.test.protocol_autogen`
+   in Debug; any under-declaration aborts the relevant `*Tests` suite.
+
+## Remaining Phase-1 migrations — turnkey (footprints analysed, verdicts fixed)
+
+STATIC (derivable from tx body):
+- CredentialCreate → `credential(sfSubject, src, sfCredentialType)` + `account(sfSubject)`
+- MPTokenIssuanceCreate → `mptIssuance(tx.getSeqValue(), src)`
+- CheckCreate → `check(src, tx.getSeqValue())` + `account(sfDestination)`
+- PaymentChannelCreate → `payChan(src, sfDestination, tx.getSeqValue())` + `account(sfDestination)`
+- Clawback → `account(src)` + `account(sfHolder)` + (IOU: `line(holder, issuer, ccy)`;
+  MPT: `mptIssuance(id)` + `mptoken(id, holder)`)
+- EscrowCreate (XRP) → `escrow(src, tx.getSeqValue())` + `account(sfDestination)`
+  (IOU/MPT variant: + issuer account + `line(src,issuer)` / mpt objects, or keep global)
+
+STATIC_WITH_SNAPSHOT (read the object SLE in `base` to resolve owners):
+- CredentialAccept → `credential(src, sfIssuer, type)` + `account(sfIssuer)`
+- CredentialDelete → `credential(subject|src, issuer|src, type)` + `account(issuer)` + `account(subject)`
+- MPTokenIssuanceDestroy → `mptIssuance(id)` + `account(issuer-from-SLE)`
+- MPTokenIssuanceSet → `mptIssuance(id)` or `mptoken(id, sfHolder)` (+ `permissionedDomain(sfDomainID)`)
+- MPTokenAuthorize → `mptIssuance(id)` + `account(holder)` + `mptoken(id, holder)` (holder = src or sfHolder)
+- CheckCancel → `check(sfCheckID)` + `account(check.Account)` + `account(check.Destination)`
+- PaymentChannelFund → `payChan(sfChannel)` + `account(chan.Account)` + `account(chan.Destination)`
+- PaymentChannelClaim → `payChan(sfChannel)` + `account(chan.Account)` + `account(chan.Destination)`
+  + `depositPreauth(chan.Destination, src)` + each `sfCredentialIDs` key
+- EscrowFinish (XRP; else global) → `escrow(sfOwner, sfOfferSequence)` + escrow.Account + escrow.Destination
+  + `depositPreauth(dst, src)` + each `sfCredentialIDs` key
+- EscrowCancel (XRP; else global) → `escrow(sfOwner, sfOfferSequence)` + escrow.Account + escrow.Destination
+
+## DYNAMIC — must stay `touchesGlobal` in v1 (reason)
+- OfferCreate, OfferCancel — shared book directory + offer crossing modifies
+  counterparty accounts not in the tx.
+- All AMM* (Create/Deposit/Withdraw/Vote/Bid/Delete/Clawback) — pool
+  pseudo-account + crossing.
+- Payment with paths / cross-currency / IOU / MPT, CheckCash — flow engine
+  (unbounded trustline/offer traversal).
+- All NFToken* — NFT page-chain `succ()` search (Mint/Burn/Modify/AcceptOffer)
+  and shared NFT offer directories (CreateOffer/CancelOffer).
+- AccountDelete — deletes every src-owned object (unbounded footprint).
+- Batch — meta-transaction; footprint is the union of its inner txs.
+- Vault*, LoanBroker*/Loan* (lending), XChain*/bridge — pseudo-accounts and
+  cross-chain/compound state; unaudited dynamic footprints.
+- PermissionedDomainSet — references arbitrary credential-issuer accounts in
+  `sfAcceptedCredentials`; keep global pending a deeper audit.
+- Change (SetFee/EnableAmendment/UNLModify), LedgerStateFix — pseudo/global.
+
+## Phase 3 core — BUILT & verified
+- `applyScheduled` (`Schedule.h`/`.cpp`): schedules a tx set, applies each
+  independent group in an isolated `OpenView` over the immutable closed
+  snapshot, and merges the disjoint write-sets into the target ledger.
+- `src/tests/libxrpl/tx/ScheduledApply.cpp`: the **serial-vs-scheduled
+  differential** — both ledgers built by the test itself (bypassing the
+  canonicalising `TxTest::close`), asserting a byte-identical, non-trivial
+  account-state root. This is the determinism-critical contract; it passes.
+
+## Phase 3 — threaded executor + server integration — BUILT
+- **Threaded execution.** `applyScheduled(..., unsigned workers)` applies the
+  independent groups across a thread pool (each group in its own view over the
+  immutable closed snapshot; disjoint write-sets merged sequentially in fixed
+  group order). `ScheduledApply.ThreadedMatchesSerialAcrossManyGroups` runs 12
+  groups across 8 threads, 8 iterations, each byte-identical to serial.
+- **Server integration.** `BuildLedger.cpp::applyTransactions` has a flag-gated
+  branch (`XRPL_PARALLEL_APPLY`, default OFF → unchanged default behaviour) that
+  schedules the canonical set and applies it via `applyScheduled`, merging into
+  the close `OpenView`. Works because `Application` *is-a* `ServiceRegistry` and
+  `OpenView` *is-a* `TxsRawView`. Compiles under `xrpld=ON`.
+  `XRPL_PARALLEL_APPLY_WORKERS` overrides the worker count
+  (default `clamp(cores-2, 2, 8)`).
+
+## Network determinism test via xrpld-lab (the live oracle)
+A local multi-validator network is itself a determinism check: if parallel apply
+were non-deterministic, validators would compute different ledger hashes and
+fail to validate. Procedure:
+1. Build: `cmake --build build --target xrpld` (xrpld=ON).
+2. `cp build/xrpld /Users/infinityworks/projects/xrplf/xrpld-lab/xrpld`
+3. `cd xrpld-lab && xrpld-lab create:network --protocol xrpl --local --num_validators 3 --num_peers 1 --genesis True`
+4. `export XRPL_PARALLEL_APPLY=1` then run the cluster's `start.sh` (the env var
+   propagates to all `nohup ./xrpld` children).
+5. Push disjoint-payment load at `ws://127.0.0.1:6016`.
+6. Assert all validators agree on `ledger_hash`/`account_hash` each round (e.g.
+   poll `server_info`/`ledger` across nodes). Divergence ⇒ a determinism bug.
+
+## Network determinism test — RUN & PASSED (xrpld-lab, 3 validators)
+Built the full `xrpld` binary from this branch and ran a 3-validator local network
+via xrpld-lab with `XRPL_PARALLEL_APPLY=1` (4 workers) on every node:
+- Consensus advanced normally (proposers=2, ~2s converge), parallel path active
+  on all nodes (logs: `Parallel apply: N applied across K group(s)`).
+- Under disjoint-payment load, every validator independently scheduled identical
+  groups — `5 applied across 5 group(s)`, `10 across 1` (same-source funding) —
+  and produced **byte-identical `ledger_hash` AND `account_hash` on every ledger**.
+- This is the within-run determinism oracle: 3 independent validators each
+  scheduling + thread-pool-applying the same tx sets agreed on every state root.
+  A nondeterministic apply would have diverged and stalled validation. It didn't.
+
+(The earlier standalone cross-restart hash comparison is an INVALID method —
+two pure-serial runs also differ because standalone ledger composition varies
+with submit/accept timing. The multi-validator within-run agreement above is the
+valid test; the unit `ScheduledApply` differential is the controlled complement.)
+
+## Measured speedup (Release, apply engine in isolation) — MODEST, overhead-bound
+`ScheduledApply.ThroughputBenchmark`, 400 disjoint payments, best of 5, Release
+(NDEBUG → access-set assertion + instrumentation compiled out):
+
+| workers | us/tx | speedup |
+|---|---|---|
+| 1 | 35.4 | 1.00x |
+| 2 | 30.4 | 1.17x |
+| 4 | 24.9 | 1.42x (peak) |
+| 8 | 35.2 | 1.01x (regressed) |
+
+Honest reading — this is NOT the headline 5–10x:
+1. **Overhead-bound.** The current engine spawns `std::thread`s *per ledger* and
+   builds an `OpenView` per group with a sequential merge. For cheap payments
+   (~35 us/tx serial) that fixed overhead rivals the work, so it peaks ~1.4x at
+   4 workers and goes net-negative by 8. A persistent thread pool + lower
+   per-group overhead is required to scale; the per-ledger spawn is the first
+   thing to fix.
+2. **Apply isn't the payment bottleneck.** ~35 us/tx serial ≈ 28k apply/s on one
+   core — far above the ~159 TPS network ceiling. For payment-dominated load the
+   ceiling is elsewhere in the pipeline (consensus, relay, admission), so
+   parallelizing *apply* alone yields limited end-to-end gain. The plan's big
+   numbers require parallelizing the EXPENSIVE transactors (AMM/DEX/paths) — and
+   those are exactly the ones still `touchesGlobal` (serial) in v1.
+3. **Worker default needs rethinking.** `clamp(cores-2, 2, 8)` over-threads this
+   workload; ~4 was best here. Tune per measurement, not a fixed default.
+
+Net: the parallelism is real and deterministic (verified), but v1's economic case
+is weak — modest payment speedup, with the large wins gated behind parallelizing
+the dynamic-footprint transactors and a lower-overhead executor.
+
+## What remains genuinely uncertifiable in a coding run
+- **Network/production certification.** A green lab run is strong evidence but
+  not proof. Shipping parallel apply to mainnet still needs: ThreadSanitizer-clean
+  runs, adversarial scheduling (Antithesis), and a 12-month mainnet-replay
+  differential CI gate (hundreds of GB, out-of-repo). A determinism bug forks the
+  network, so these are non-negotiable before the flag becomes an amendment.
+- **Phase 5 (amendment).** `ParallelApply` amendment + validator governance vote.
+- **Productionization of the flag** (Config stanza / amendment gate instead of an
+  env var) and faithful failed/retry-set parity for adversarial (invalid-tx)
+  workloads — the current branch clears the set after a successful grouped apply,
+  which is correct for valid load but not yet a full match of the serial path's
+  failure bookkeeping.
+- **Phase 4 (testnet load).** Operational: testnet + load harness + weeks of
+  runtime to hit the ≥1000 TPS target. Cannot be done from the repo.
+- **Phase 5 (amendment).** `ParallelApply` amendment + governance vote.
+
+## How to verify locally
+Debug build, `-Dtests=ON -Dxrpld=OFF`; build `xrpl.test.tx` and
+`xrpl.test.protocol_autogen`; run both. The DEBUG assertion validates every
+migrated transactor against real apply tests. `XRPL_ACCESS_AUDIT=1` logs the
+measured footprint of `touchesGlobal` transactors (Phase-1.5 audit data).