Merge branch 'dev' into emit-atomic-v3

hook/error.h

@@ -48,5 +48,6 @@
 #define TOO_MANY_STATE_MODIFICATIONS -44
 #define TOO_MANY_NAMESPACES -45
 #define TOO_LITTLE_ENTROPY -46
+#define ALREADY_IN_SUBLEDGER -47
 #define HOOK_ERROR_CODES
-#endif //HOOK_ERROR_CODES
+#endif  // HOOK_ERROR_CODES

hook/extern.h

@@ -345,6 +345,23 @@ dice(uint32_t sides);
 extern int64_t
 util_random(uint32_t write_ptr, uint32_t write_len);
 
+extern int64_t
+emit_atomic(
+    uint32_t write_ptr,
+    uint32_t write_len,
+    uint32_t read_ptr,
+    uint32_t read_len);
+
+extern int64_t
+prepare_atomic(
+    uint32_t write_ptr,
+    uint32_t write_len,
+    uint32_t read_ptr,
+    uint32_t read_len);
+
+extern int64_t
+subledger_slot(uint32_t slot_into, uint32_t read_ptr, uint32_t read_len);
+
 #ifdef __cplusplus
 }
 #endif

hook/generate_error.sh

@@ -55,4 +55,4 @@ sed -n '/enum class hook_return_code/,/};/p' "$ENUM_FILE" |
         }
     '
 echo '#define HOOK_ERROR_CODES'
-echo '#endif //HOOK_ERROR_CODES'
+echo '#endif  // HOOK_ERROR_CODES'

hook/sfcodes.h

@@ -306,6 +306,7 @@
 #define sfRemark ((14U << 16U) + 97U)
 #define sfHighReward ((14U << 16U) + 98U)
 #define sfLowReward ((14U << 16U) + 99U)
+#define sfSubledgerTransaction ((14U << 16U) + 100U)
 #define sfSigners ((15U << 16U) + 3U)
 #define sfSignerEntries ((15U << 16U) + 4U)
 #define sfTemplate ((15U << 16U) + 5U)
@@ -335,3 +336,4 @@
 #define sfActiveValidators ((15U << 16U) + 95U)
 #define sfGenesisMints ((15U << 16U) + 96U)
 #define sfRemarks ((15U << 16U) + 97U)
+#define sfSubledger ((15U << 16U) + 98U)

hook/tx_flags.h

@@ -115,3 +115,11 @@ enum AMMClawbackFlags : uint32_t {
 enum BridgeModifyFlags : uint32_t {
     tfClearAccountCreateAmount = 0x00010000,
 };
+
+enum SubledgerTransactionFlags : uint32_t {
+    // the transaction's Hooks, or those of a transaction applied before it
+    // in the same subledger, drew from the RNG: the outcome the emitting
+    // Hook saw was drawn from a shadow RNG and may differ from the one
+    // committed
+    stfIndeterminate = 0x00000001,
+};

include/xrpl/hook/Enum.h

@@ -18,6 +18,7 @@
 #define featureRNG "1"
 #define fix20250131 "1"
 #define fixGuardDepth32 "1"
+#define featureAtomicEmit "1"
 namespace hook_api {
 struct Rules
 {
@@ -387,8 +388,10 @@ enum class hook_return_code : int64_t {
     TOO_MANY_STATE_MODIFICATIONS = -44,  // more than 256 modified state
                                          // entires in the combined hook chains
     TOO_MANY_NAMESPACES = -45,
-    TOO_LITTLE_ENTROPY = -46  // featureRNG: no seed for this ledger, or too few
-                              // validators contributed to it
+    TOO_LITTLE_ENTROPY = -46,   // featureRNG: no seed for this ledger, or too
+                                // few validators contributed to it
+    ALREADY_IN_SUBLEDGER = -47  // emit_atomic called by a hook executing
+                                // inside a subledger (an emit_atomic txn)
 };
 
 enum class ExitType : uint8_t {
@@ -403,6 +406,10 @@ const uint8_t max_slots = 255;
 const uint8_t max_nonce = 255;
 const uint16_t max_rng_draws = 256;  // featureRNG, per Hook execution
 const uint8_t max_emit = 255;
+// maximum number of emit_atomic attempts per outer transaction, across all
+// hook executions. Every attempt that reaches execution counts, whether or not
+// the txn ends up in the subledger.
+const uint8_t max_atomic_emit = 8;
 const uint8_t max_params = 16;
 const double fee_base_multiplier = 1.1f;
 

include/xrpl/hook/hook_api.macro

@@ -306,4 +306,16 @@ HOOK_API_DEFINITION(
 HOOK_API_DEFINITION(
   int64_t, util_random, (uint32_t write_ptr, uint32_t write_len),
   featureRNG)
+
+HOOK_API_DEFINITION(
+  int64_t, emit_atomic, (uint32_t write_ptr, uint32_t write_len, uint32_t read_ptr, uint32_t read_len),
+  featureAtomicEmit)
+
+HOOK_API_DEFINITION(
+  int64_t, prepare_atomic, (uint32_t write_ptr, uint32_t write_len, uint32_t read_ptr, uint32_t read_len),
+  featureAtomicEmit)
+
+HOOK_API_DEFINITION(
+  int64_t, subledger_slot, (uint32_t slot_into, uint32_t read_ptr, uint32_t read_len),
+  featureAtomicEmit)
 // clang-format on

include/xrpl/protocol/TER.h

@@ -367,6 +367,7 @@ enum TECcodes : TERUnderlyingType {
     tecARRAY_TOO_LARGE = 197,
     tecLOCKED = 198,
     tecBAD_CREDENTIALS = 199,
+    tecSUBLEDGER_DIVERGED = 200,
     tecLAST_POSSIBLE_ENTRY = 255,
 };
 

include/xrpl/protocol/TxFlags.h

@@ -274,6 +274,15 @@ enum BridgeModifyFlags : uint32_t {
     tfClearAccountCreateAmount = 0x00010000,
 };
 constexpr std::uint32_t tfBridgeModifyMask = ~(tfUniversal | tfClearAccountCreateAmount);
+
+// SubledgerTransaction flags (featureAtomicEmit, metadata only):
+enum SubledgerTransactionFlags : uint32_t {
+    // the transaction's Hooks, or those of a transaction applied before it
+    // in the same subledger, drew from the RNG: the outcome the emitting
+    // Hook saw was drawn from a shadow RNG and may differ from the one
+    // committed
+    stfIndeterminate = 0x00000001,
+};
 // clang-format on
 
 }  // namespace ripple

include/xrpl/protocol/TxMeta.h

@@ -172,6 +172,33 @@ public:
         return *mRandomData;
     }
 
+    /** The Subledger: the atomically emitted (emit_atomic) transactions
+        that were applied as part of this one, each as a
+        SubledgerTransaction { EmittedTxnID, EmittedTxn, TransactionMetaData }.
+        This transaction's AffectedNodes already include their effects.
+    */
+    STArray const&
+    getSubledger() const
+    {
+        // unlike the Hook arrays this is read by callers that only know the
+        // transaction had hooks (explorers, subledger_slot on a parent that
+        // never emitted): never dereference an empty optional
+        static STArray const empty{sfSubledger};
+        return mSubledger ? *mSubledger : empty;
+    }
+
+    void
+    setSubledger(STArray const& subledger)
+    {
+        mSubledger = subledger;
+    }
+
+    bool
+    hasSubledger() const
+    {
+        return static_cast<bool>(mSubledger);
+    }
+
     STAmount
     getDeliveredAmount() const
     {
@@ -197,6 +224,7 @@ private:
     std::optional<STArray> mHookExecutions;
     std::optional<STArray> mHookEmissions;
     std::optional<uint256> mRandomData;
+    std::optional<STArray> mSubledger;
 
     STArray mNodes;
 };

include/xrpl/protocol/detail/features.macro

@@ -31,9 +31,10 @@
 
 // Add new amendments to the top of this list.
 // Keep it sorted in reverse chronological order.
-// If you add an amendment here, then do not forget to increment `numFeatures`
-// in include/xrpl/protocol/Feature.h.
+// `numFeatures` in include/xrpl/protocol/Feature.h is derived from this file
+// by re-including it, so adding an amendment here needs no other change.
 
+XRPL_FEATURE(AtomicEmit,                 Supported::yes, VoteBehavior::DefaultNo)
 XRPL_FEATURE(RNG,                        Supported::yes, VoteBehavior::DefaultNo)
 XRPL_FEATURE(PWALoader,                  Supported::yes, VoteBehavior::DefaultNo)
 XRPL_FIX    (20261005,                   Supported::yes, VoteBehavior::DefaultYes)

include/xrpl/protocol/detail/sfields.macro

@@ -401,6 +401,7 @@ UNTYPED_SFIELD(sfGenesisMint,            OBJECT,    96)
 UNTYPED_SFIELD(sfRemark,                 OBJECT,    97)
 UNTYPED_SFIELD(sfHighReward,             OBJECT,    98)
 UNTYPED_SFIELD(sfLowReward,              OBJECT,    99)
+UNTYPED_SFIELD(sfSubledgerTransaction,   OBJECT,   100)
 
 // array of objects (common)
 // ARRAY/1 is reserved for end of array
@@ -439,3 +440,4 @@ UNTYPED_SFIELD(sfImportVLKeys,           ARRAY,     94)
 UNTYPED_SFIELD(sfActiveValidators,       ARRAY,     95)
 UNTYPED_SFIELD(sfGenesisMints,           ARRAY,     96)
 UNTYPED_SFIELD(sfRemarks,                ARRAY,     97)
+UNTYPED_SFIELD(sfSubledger,              ARRAY,     98)

src/libxrpl/protocol/InnerObjectFormats.cpp

@@ -87,6 +87,15 @@ InnerObjectFormats::InnerObjectFormats()
          {sfEmittedTxnID, soeREQUIRED},
          {sfEmitNonce, soeOPTIONAL}});
 
+    // one atomically emitted (emit_atomic) transaction in the Subledger
+    // metadata of the transaction whose hook emitted it
+    add(sfSubledgerTransaction.jsonName,
+        sfSubledgerTransaction.getCode(),
+        {{sfEmittedTxnID, soeREQUIRED},
+         {sfFlags, soeOPTIONAL},
+         {sfEmittedTxn, soeREQUIRED},
+         {sfTransactionMetaData, soeREQUIRED}});
+
     add(sfHook.jsonName,
         sfHook.getCode(),
         {{sfHookHash, soeOPTIONAL},

src/libxrpl/protocol/TER.cpp

@@ -124,6 +124,7 @@ transResults()
         MAKE_ERROR(tecARRAY_TOO_LARGE,               "Array is too large."),
         MAKE_ERROR(tecLOCKED,                        "Fund is locked."),
         MAKE_ERROR(tecBAD_CREDENTIALS,               "Bad credentials."),
+        MAKE_ERROR(tecSUBLEDGER_DIVERGED,            "The real subledger could not be reconciled with the one the strong Hooks saw."),
 
         MAKE_ERROR(tefALREADY,                     "The exact transaction was already in this ledger."),
         MAKE_ERROR(tefBAD_ADD_AUTH,                "Not authorized to add account."),

src/libxrpl/protocol/TxMeta.cpp

@@ -52,6 +52,9 @@ TxMeta::TxMeta(
 
     if (obj.isFieldPresent(sfRandomData))
         setRandomData(obj.getFieldH256(sfRandomData));
+
+    if (obj.isFieldPresent(sfSubledger))
+        setSubledger(obj.getFieldArray(sfSubledger));
 }
 
 TxMeta::TxMeta(uint256 const& txid, std::uint32_t ledger, STObject const& obj)
@@ -81,6 +84,9 @@ TxMeta::TxMeta(uint256 const& txid, std::uint32_t ledger, STObject const& obj)
 
     if (obj.isFieldPresent(sfRandomData))
         setRandomData(obj.getFieldH256(sfRandomData));
+
+    if (obj.isFieldPresent(sfSubledger))
+        setSubledger(obj.getFieldArray(sfSubledger));
 }
 
 TxMeta::TxMeta(uint256 const& txid, std::uint32_t ledger, Blob const& vec)
@@ -254,6 +260,9 @@ TxMeta::getAsObject() const
     if (hasRandomData())
         metaData.setFieldH256(sfRandomData, getRandomData());
 
+    if (hasSubledger())
+        metaData.setFieldArray(sfSubledger, getSubledger());
+
     return metaData;
 }
 

src/test/app/HookAPI_test.cpp

@@ -18,12 +18,17 @@
 //==============================================================================
 #include <test/app/Import_json.h>
 #include <test/jtx.h>
+#include <test/jtx/hook.h>
 #include <xrpld/app/hook/HookAPI.h>
+#include <xrpld/app/hook/applyHook.h>
+#include <xrpld/app/tx/applySteps.h>
 #include <xrpl/basics/StringUtilities.h>
 #include <xrpl/beast/unit_test/suite.h>
 #include <xrpl/json/json_writer.h>
 #include <xrpl/protocol/SField.h>
 #include <xrpl/protocol/STAccount.h>
+#include <xrpl/protocol/TxFlags.h>
+#include <xrpl/protocol/digest.h>
 #include <limits>
 #include <tuple>
 #include <vector>
@@ -32,11 +37,44 @@ namespace ripple {
 
 namespace test {
 
+namespace {
+
+// accept()s with dice(0xFFFFFFFF) as the return code (copied from RNG_test)
+// (module
+//   (import "env" "_g" (func $_g (param i32 i32) (result i32)))
+//   (import "env" "accept" (func $accept (param i32 i32 i64) (result i64)))
+//   (import "env" "dice" (func $dice (param i32) (result i64)))
+//   (memory 1)
+//   (func $hook (param i32) (result i64)
+//     (drop (call $_g (i32.const 1) (i32.const 1)))
+//     (call $accept (i32.const 0) (i32.const 0) (call $dice (i32.const -1))))
+//   (export "memory" (memory 0))
+//   (export "hook" (func $hook)))
+static std::vector<std::uint8_t> const diceMaxWasm = {
+    0x00U, 0x61U, 0x73U, 0x6DU, 0x01U, 0x00U, 0x00U, 0x00U, 0x01U, 0x13U, 0x03U,
+    0x60U, 0x02U, 0x7FU, 0x7FU, 0x01U, 0x7FU, 0x60U, 0x03U, 0x7FU, 0x7FU, 0x7EU,
+    0x01U, 0x7EU, 0x60U, 0x01U, 0x7FU, 0x01U, 0x7EU, 0x02U, 0x22U, 0x03U, 0x03U,
+    0x65U, 0x6EU, 0x76U, 0x02U, 0x5FU, 0x67U, 0x00U, 0x00U, 0x03U, 0x65U, 0x6EU,
+    0x76U, 0x06U, 0x61U, 0x63U, 0x63U, 0x65U, 0x70U, 0x74U, 0x00U, 0x01U, 0x03U,
+    0x65U, 0x6EU, 0x76U, 0x04U, 0x64U, 0x69U, 0x63U, 0x65U, 0x00U, 0x02U, 0x03U,
+    0x02U, 0x01U, 0x02U, 0x05U, 0x03U, 0x01U, 0x00U, 0x01U, 0x07U, 0x11U, 0x02U,
+    0x06U, 0x6DU, 0x65U, 0x6DU, 0x6FU, 0x72U, 0x79U, 0x02U, 0x00U, 0x04U, 0x68U,
+    0x6FU, 0x6FU, 0x6BU, 0x00U, 0x03U, 0x0AU, 0x15U, 0x01U, 0x13U, 0x00U, 0x41U,
+    0x01U, 0x41U, 0x01U, 0x10U, 0x00U, 0x1AU, 0x41U, 0x00U, 0x41U, 0x00U, 0x41U,
+    0x7FU, 0x10U, 0x02U, 0x10U, 0x01U, 0x0BU,
+};
+
+}  // namespace
+
 class HookAPI_test : public beast::unit_test::suite
 {
 private:
     ApplyContext
-    createApplyContext(jtx::Env& env, OpenView& ov, STTx const& tx)
+    createApplyContext(
+        jtx::Env& env,
+        OpenView& ov,
+        STTx const& tx,
+        ApplyFlags flags = tapNONE)
     {
         ApplyContext applyCtx{
             env.app(),
@@ -44,7 +82,7 @@ private:
             tx,
             tesSUCCESS,
             env.current()->fees().base,
-            tapNONE,
+            flags,
             env.journal};
         return applyCtx;
     }
@@ -154,6 +192,875 @@ public:
                 api.emit(Slice(result.value().data(), result.value().size()));
             BEAST_EXPECT(result2.has_value());
         }
+
+        if (env.current()->rules().enabled(featureAtomicEmit))
+        {
+            // prepare_atomic: the ledger window is forced to the current
+            // ledger even when the blob already carries a different one, and
+            // the result is accepted by emit_atomic (but not by emit)
+            auto hookCtx = makeStubHookContext(
+                applyCtx,
+                alice.id(),
+                alice.id(),
+                {
+                    .expected_etxn_count = 1,
+                    .result = {.isStrong = true},
+                });
+            auto& api = hookCtx.api();
+            auto tx = emitInvokeTx;  // carries FLS = seq + 1, LLS = seq + 5
+            Serializer s = tx.getSerializer();
+            auto const result = api.prepare(s.slice(), /*atomic=*/true);
+            BEAST_EXPECT(result.has_value());
+
+            SerialIter sit(Slice(result.value().data(), result.value().size()));
+            STObject st(sit, sfGeneric);
+            auto const seq = applyCtx.view().info().seq;
+            BEAST_EXPECT(st.getFieldU32(sfFirstLedgerSequence) == seq);
+            BEAST_EXPECT(st.getFieldU32(sfLastLedgerSequence) == seq);
+            BEAST_EXPECT(st.getFieldAmount(sfFee) > XRPAmount(0));
+            BEAST_EXPECT(st.getFieldU32(sfSequence) == 0);
+            BEAST_EXPECT(st.isFieldPresent(sfEmitDetails));
+
+            Slice const prepared(result.value().data(), result.value().size());
+            BEAST_EXPECT(api.emit(prepared).error() == EMISSION_FAILURE);
+            BEAST_EXPECT(api.emit(prepared, /*atomic=*/true).has_value());
+        }
+    }
+
+    void
+    test_emit_atomic_subledger(FeatureBitset features)
+    {
+        testcase("Test emit_atomic subledger");
+        using namespace jtx;
+        using namespace hook_api;
+
+        if (!features[featureAtomicEmit])
+            return;
+
+        auto const alice = Account{"alice"};
+        auto const bob = Account{"bob"};
+
+        Env env{*this, features};
+        env.fund(XRP(10000), alice, bob);
+        env.close();
+
+        // a closed base, so that committing generates metadata
+        auto const closed = env.closed();
+
+        // the transaction the (stub) hook is executing for
+        auto const makeOtxn = [&](XRPAmount fee) {
+            return STTx(ttINVOKE, [&](STObject& obj) {
+                obj[sfAccount] = alice.id();
+                obj[sfSequence] = env.seq(alice);
+                obj[sfSigningPubKey] = Slice{};
+                obj[sfFee] = fee;
+            });
+        };
+
+        // an atomically emitted payment from the hook account to bob
+        auto const makePayment =
+            [&](OpenView const& ov, STTx const& otxn, XRPAmount amount) {
+                return STTx(ttPAYMENT, [&](STObject& obj) {
+                    obj[sfAccount] = alice.id();
+                    obj[sfDestination] = bob.id();
+                    obj[sfAmount] = STAmount{amount};
+                    obj[sfSequence] = 0;
+                    obj[sfSigningPubKey] = Slice{};
+                    obj[sfFirstLedgerSequence] = ov.seq();
+                    obj[sfLastLedgerSequence] = ov.seq();
+                    obj[sfFee] = env.closed()->fees().base;
+
+                    auto& emitDetails = obj.peekFieldObject(sfEmitDetails);
+                    emitDetails[sfEmitGeneration] = 1;
+                    emitDetails[sfEmitBurden] = 1;
+                    emitDetails[sfEmitParentTxnID] = otxn.getTransactionID();
+                    emitDetails[sfEmitNonce] = uint256();
+                    emitDetails[sfEmitHookHash] = uint256();
+                });
+            };
+
+        auto const bobBalance = [&](ReadView const& view) {
+            return view.read(keylet::account(bob.id()))
+                ->getFieldAmount(sfBalance);
+        };
+
+        {
+            // executed immediately, visible to the hook, inspectable, and
+            // committed as part of the transaction
+            STTx const otxn = makeOtxn(XRPAmount{10});
+            OpenView ov(&*closed);
+            ApplyContext applyCtx = createApplyContext(env, ov, otxn);
+            auto hookCtx = makeStubHookContext(
+                applyCtx,
+                alice.id(),
+                alice.id(),
+                {.expected_etxn_count = 1,
+                 .nonce_used = {{uint256(0), true}},
+                 .result = {.isStrong = true}});
+            auto& api = hookCtx.api();
+
+            STTx const pay = makePayment(ov, otxn, XRP(100).value().xrp());
+            Serializer const ser = pay.getSerializer();
+            auto const tpTrans = api.emit(ser.slice(), /*atomic=*/true);
+            BEAST_EXPECT(tpTrans.has_value());
+            if (!tpTrans)
+                return;
+
+            auto const bobBefore =
+                applyCtx.view().read(keylet::account(bob.id()));
+            BEAST_EXPECT(!applyCtx.hasSubledger());
+            BEAST_EXPECT(applyCtx.atomicEmitCount == 0);
+
+            auto const entry = api.apply_atomic(*tpTrans);
+            BEAST_EXPECT(entry.has_value());
+            if (!entry)
+                return;
+            // apply_atomic() did the emission bookkeeping itself
+            BEAST_EXPECT(applyCtx.atomicEmitCount == 1);
+            BEAST_EXPECT(
+                hookCtx.nonce_consumed.count(uint256(0)) == 1 &&
+                hookCtx.nonce_consumed[uint256(0)] == true);
+            BEAST_EXPECT(applyCtx.hasSubledger());
+            BEAST_EXPECT(applyCtx.subledgerEntries().size() == 1);
+
+            STObject const& e = **entry;
+            BEAST_EXPECT(
+                e.getFieldH256(sfEmittedTxnID) == pay.getTransactionID());
+            auto const& innerMeta =
+                e.peekAtField(sfTransactionMetaData).downcast<STObject>();
+            BEAST_EXPECT(
+                innerMeta.getFieldU8(sfTransactionResult) ==
+                TERtoInt(tesSUCCESS));
+            BEAST_EXPECT(innerMeta.getFieldU32(sfTransactionIndex) == 0);
+
+            // the hook, and everything applied after it, reads through the
+            // subledger
+            BEAST_EXPECT(
+                bobBalance(applyCtx.view()) ==
+                bobBefore->getFieldAmount(sfBalance) + STAmount{XRP(100)});
+            BEAST_EXPECT(
+                bobBalance(ov) == bobBefore->getFieldAmount(sfBalance));
+
+            // the subledger's changes (alice and bob) count towards the
+            // oversize cap even though the working view itself is empty
+            BEAST_EXPECT(applyCtx.size() == 2);
+
+            // the provisional metadata already lists the Subledger
+            {
+                auto const proMeta = applyCtx.generateProvisionalMeta();
+                BEAST_EXPECT(
+                    proMeta.hasSubledger() &&
+                    proMeta.getSubledger().size() == 1);
+            }
+
+            // subledger_slot by txid and by index
+            BEAST_EXPECT(
+                api.subledger_slot(0, pay.getTransactionID(), 0).has_value());
+            BEAST_EXPECT(api.subledger_slot(0, std::nullopt, 0).has_value());
+            BEAST_EXPECT(
+                api.subledger_slot(0, std::nullopt, 1).error() == DOESNT_EXIST);
+            BEAST_EXPECT(
+                api.subledger_slot(0, uint256(7), 0).error() == DOESNT_EXIST);
+
+            // the same nonce cannot be used again
+            BEAST_EXPECT(
+                api.emit(ser.slice(), /*atomic=*/true).error() ==
+                EMISSION_FAILURE);
+
+            // commit: only the transaction enters the ledger, its
+            // AffectedNodes are the net change, threaded to it, and the
+            // payment is in its Subledger
+            auto const bobPrevTxn = bobBefore->getFieldH256(sfPreviousTxnID);
+            auto txMeta = applyCtx.apply(tesSUCCESS);
+            BEAST_EXPECT(txMeta.has_value());
+            BEAST_EXPECT(ov.txCount() == 1);
+            BEAST_EXPECT(ov.txExists(otxn.getTransactionID()));
+            BEAST_EXPECT(!ov.txExists(pay.getTransactionID()));
+            if (txMeta)
+            {
+                BEAST_EXPECT(
+                    txMeta->hasSubledger() &&
+                    txMeta->getSubledger().size() == 1);
+                bool found = false;
+                for (auto const& node : txMeta->getNodes())
+                {
+                    if (node.getFieldH256(sfLedgerIndex) !=
+                        keylet::account(bob.id()).key)
+                        continue;
+                    found = true;
+                    BEAST_EXPECT(node.getFName() == sfModifiedNode);
+                    BEAST_EXPECT(
+                        node.getFieldH256(sfPreviousTxnID) == bobPrevTxn);
+                }
+                BEAST_EXPECT(found);
+
+                // the Subledger survives a serialisation round trip
+                Serializer s;
+                txMeta->addRaw(s, tesSUCCESS, 0);
+                TxMeta const parsed(
+                    otxn.getTransactionID(), ov.seq(), s.peekData());
+                BEAST_EXPECT(
+                    parsed.hasSubledger() && parsed.getSubledger().size() == 1);
+            }
+            BEAST_EXPECT(
+                bobBalance(ov) ==
+                bobBefore->getFieldAmount(sfBalance) + STAmount{XRP(100)});
+            BEAST_EXPECT(
+                ov.read(keylet::account(bob.id()))
+                    ->getFieldH256(sfPreviousTxnID) == otxn.getTransactionID());
+        }
+
+        {
+            // abandoned: discard() (rollback, or any tec) drops the
+            // subledger
+            STTx const otxn = makeOtxn(XRPAmount{10});
+            OpenView ov(&*closed);
+            ApplyContext applyCtx = createApplyContext(env, ov, otxn);
+            auto hookCtx = makeStubHookContext(
+                applyCtx,
+                alice.id(),
+                alice.id(),
+                {.expected_etxn_count = 1,
+                 .nonce_used = {{uint256(0), true}},
+                 .result = {.isStrong = true}});
+            auto& api = hookCtx.api();
+
+            STTx const pay = makePayment(ov, otxn, XRP(100).value().xrp());
+            Serializer const ser = pay.getSerializer();
+            auto const tpTrans = api.emit(ser.slice(), /*atomic=*/true);
+            BEAST_EXPECT(tpTrans.has_value());
+            if (tpTrans)
+            {
+                BEAST_EXPECT(api.apply_atomic(*tpTrans).has_value());
+                BEAST_EXPECT(applyCtx.hasSubledger());
+                applyCtx.discard();
+                BEAST_EXPECT(!applyCtx.hasSubledger());
+                BEAST_EXPECT(applyCtx.subledgerEntries().empty());
+                BEAST_EXPECT(
+                    bobBalance(applyCtx.view()) == bobBalance(*closed));
+                BEAST_EXPECT(
+                    api.subledger_slot(0, std::nullopt, 0).error() ==
+                    DOESNT_EXIST);
+            }
+        }
+
+        {
+            // the transaction's fee is charged after its subledger: an inner
+            // txn that would leave it unable to pay is refused
+            auto const aliceBalance = env.balance(alice).value().xrp();
+            STTx const otxn = makeOtxn(aliceBalance - XRP(100).value().xrp());
+            OpenView ov(&*closed);
+            ApplyContext applyCtx = createApplyContext(env, ov, otxn);
+            auto hookCtx = makeStubHookContext(
+                applyCtx,
+                alice.id(),
+                alice.id(),
+                {.expected_etxn_count = 1,
+                 .nonce_used = {{uint256(0), true}},
+                 .result = {.isStrong = true}});
+            auto& api = hookCtx.api();
+
+            STTx const pay = makePayment(ov, otxn, XRP(1000).value().xrp());
+            Serializer const ser = pay.getSerializer();
+            auto const tpTrans = api.emit(ser.slice(), /*atomic=*/true);
+            BEAST_EXPECT(tpTrans.has_value());
+            if (tpTrans)
+            {
+                BEAST_EXPECT(
+                    api.apply_atomic(*tpTrans).error() == EMISSION_FAILURE);
+                BEAST_EXPECT(applyCtx.subledgerEntries().empty());
+                BEAST_EXPECT(
+                    bobBalance(applyCtx.view()) == bobBalance(*closed));
+            }
+        }
+    }
+
+    void
+    test_subledger_merge()
+    {
+        testcase("Test subledger entry merge");
+        using namespace jtx;
+
+        auto const alice = Account{"alice"};
+        auto const bob = Account{"bob"};
+
+        auto const make = [&](XRPAmount balance, std::uint32_t ownerCount) {
+            auto sle = std::make_shared<SLE>(keylet::account(alice.id()));
+            sle->setAccountID(sfAccount, alice.id());
+            sle->setFieldAmount(sfBalance, STAmount{balance});
+            sle->setFieldU32(sfOwnerCount, ownerCount);
+            sle->setFieldU32(sfSequence, 5);
+            return sle;
+        };
+
+        auto const before = make(XRPAmount{100}, 0);
+
+        {
+            // disjoint fields: ours from ours, the rest from theirs
+            auto const ours = make(XRPAmount{100}, 1);
+            auto const theirs = make(XRPAmount{150}, 0);
+            auto const merged = subledger::mergeEntry(*before, *ours, *theirs);
+            BEAST_EXPECT(merged);
+            if (merged)
+            {
+                BEAST_EXPECT(
+                    merged->getFieldAmount(sfBalance) ==
+                    STAmount{XRPAmount{150}});
+                BEAST_EXPECT(merged->getFieldU32(sfOwnerCount) == 1);
+                BEAST_EXPECT(merged->getFieldU32(sfSequence) == 5);
+            }
+        }
+
+        {
+            // the same field changed on both sides conflicts
+            auto const ours = make(XRPAmount{90}, 0);
+            auto const theirs = make(XRPAmount{150}, 0);
+            BEAST_EXPECT(!subledger::mergeEntry(*before, *ours, *theirs));
+        }
+
+        {
+            // ... unless theirs did not change it
+            auto const ours = make(XRPAmount{90}, 0);
+            auto const theirs = make(XRPAmount{100}, 0);
+            auto const merged = subledger::mergeEntry(*before, *ours, *theirs);
+            BEAST_EXPECT(
+                merged &&
+                merged->getFieldAmount(sfBalance) == STAmount{XRPAmount{90}});
+        }
+
+        {
+            // optional fields: added by ours, removed by ours
+            auto const withKey = make(XRPAmount{100}, 0);
+            withKey->setAccountID(sfRegularKey, bob.id());
+
+            auto const ours = make(XRPAmount{100}, 0);
+            ours->setFieldVL(sfDomain, Slice{"x", 1});
+            auto const theirs = make(XRPAmount{150}, 0);
+            auto const merged = subledger::mergeEntry(*before, *ours, *theirs);
+            BEAST_EXPECT(merged && merged->isFieldPresent(sfDomain));
+
+            // removed by ours while theirs kept it
+            auto const theirsKey = make(XRPAmount{150}, 0);
+            theirsKey->setAccountID(sfRegularKey, bob.id());
+            auto const removed = subledger::mergeEntry(
+                *withKey, *make(XRPAmount{100}, 0), *theirsKey);
+            BEAST_EXPECT(removed && !removed->isFieldPresent(sfRegularKey));
+
+            // removed by ours while theirs changed it
+            auto const theirsOther = make(XRPAmount{100}, 0);
+            theirsOther->setAccountID(sfRegularKey, alice.id());
+            BEAST_EXPECT(!subledger::mergeEntry(
+                *withKey, *make(XRPAmount{100}, 0), *theirsOther));
+        }
+    }
+
+    void
+    test_emit_atomic_rng(FeatureBitset features)
+    {
+        testcase("Test emit_atomic RNG (shadow and real subledgers)");
+        using namespace jtx;
+        using namespace hook_api;
+
+        if (!features[featureAtomicEmit] || !features[featureRNG])
+            return;
+
+        auto const alice = Account{"alice"};
+        auto const bob = Account{"bob"};
+        auto const carol = Account{"carol"};
+
+        // in standalone mode a validator contributes to every ledger it closes
+        Env env{*this, envconfig(validator, ""), features};
+        env.fund(XRP(10000), alice, bob, carol);
+        env.close();
+        // bob draws from the RNG on every transaction he is a stake holder in
+        env(hook(bob, {{hso(diceMaxWasm)}}, 0), fee(XRP(100)));
+        env.close();
+        env.close();
+
+        auto const closed = env.closed();
+        auto const seedSle = closed->read(keylet::random());
+        if (!BEAST_EXPECT(
+                seedSle &&
+                seedSle->getFieldU32(sfLedgerSequence) == closed->seq()))
+            return;
+        uint256 const seed = seedSle->getFieldH256(sfRandomData);
+
+        auto const step = [](uint256 const& s) {
+            return sha512Half(std::uint32_t{0x524E4749}, s);
+        };
+        auto const randomState = [](ReadView const& view) {
+            return view.read(keylet::random())->getFieldH256(sfRandomData);
+        };
+        auto const hookCode = [](STObject const& entry) -> std::uint64_t {
+            auto const& meta =
+                entry.peekAtField(sfTransactionMetaData).downcast<STObject>();
+            if (!meta.isFieldPresent(sfHookExecutions))
+                return 0;
+            auto const& execs = meta.getFieldArray(sfHookExecutions);
+            return execs.empty() ? 0 : execs[0].getFieldU64(sfHookReturnCode);
+        };
+        auto const noRngIn = [](STObject const& entry) {
+            auto const& meta =
+                entry.peekAtField(sfTransactionMetaData).downcast<STObject>();
+            if (meta.isFieldPresent(sfRandomData))
+                return false;
+            for (auto const& node : meta.getFieldArray(sfAffectedNodes))
+                if (node.getFieldU16(sfLedgerEntryType) == ltRANDOM)
+                    return false;
+            return true;
+        };
+
+        auto const makeOtxn = [&]() {
+            return STTx(ttINVOKE, [&](STObject& obj) {
+                obj[sfAccount] = alice.id();
+                obj[sfSequence] = env.seq(alice);
+                obj[sfSigningPubKey] = Slice{};
+                obj[sfFee] = XRPAmount{10};
+            });
+        };
+        auto const makePayment = [&](OpenView const& ov,
+                                     STTx const& otxn,
+                                     Account const& to,
+                                     std::uint8_t nonce) {
+            return std::make_shared<STTx const>(ttPAYMENT, [&](STObject& obj) {
+                obj[sfAccount] = alice.id();
+                obj[sfDestination] = to.id();
+                obj[sfAmount] = STAmount{XRP(100).value().xrp()};
+                obj[sfSequence] = 0;
+                obj[sfSigningPubKey] = Slice{};
+                obj[sfFirstLedgerSequence] = ov.seq();
+                obj[sfLastLedgerSequence] = ov.seq();
+                // generous: covers bob's hook
+                obj[sfFee] = XRP(1).value().xrp();
+
+                auto& emitDetails = obj.peekFieldObject(sfEmitDetails);
+                emitDetails[sfEmitGeneration] = 1;
+                emitDetails[sfEmitBurden] = 1;
+                emitDetails[sfEmitParentTxnID] = otxn.getTransactionID();
+                emitDetails[sfEmitNonce] = uint256(nonce);
+                emitDetails[sfEmitHookHash] = uint256();
+            });
+        };
+
+        {
+            // An inner whose Hooks draw forks the subledgers: the hook sees
+            // the shadow outcome, the real one is committed.
+            STTx const otxn = makeOtxn();
+            OpenView ov(&*closed);
+            ApplyContext applyCtx = createApplyContext(env, ov, otxn);
+
+            auto const r =
+                applyCtx.applyToSubledger(makePayment(ov, otxn, bob, 1));
+            if (!BEAST_EXPECT(r.entry))
+                return;
+            BEAST_EXPECT(isTesSuccess(r.ter));
+            BEAST_EXPECT(
+                r.entry->isFieldPresent(sfFlags) &&
+                r.entry->getFieldU32(sfFlags) == stfIndeterminate);
+            BEAST_EXPECT(noRngIn(*r.entry));
+            auto const visibleCode = hookCode(*r.entry);
+
+            // the Random object is not touched while there is a subledger
+            BEAST_EXPECT(randomState(applyCtx.view()) == seed);
+            BEAST_EXPECT(applyCtx.subledgerRngDrawn());
+
+            // a later inner that does not draw is tainted too
+            auto const r2 =
+                applyCtx.applyToSubledger(makePayment(ov, otxn, carol, 2));
+            BEAST_EXPECT(
+                r2.entry && r2.entry->isFieldPresent(sfFlags) &&
+                r2.entry->getFieldU32(sfFlags) == stfIndeterminate);
+
+            // a change this transaction makes on top of the visible
+            // subledger is merged into the real one
+            {
+                auto sle = applyCtx.view().peek(keylet::account(alice.id()));
+                sle->setFieldVL(sfDomain, Slice{"x", 1});
+                applyCtx.view().update(sle);
+            }
+
+            // end of the strong phase: continue on the real subledger
+            BEAST_EXPECT(applyCtx.concretizeSubledger());
+            BEAST_EXPECT(applyCtx.subledgerEntries().size() == 2);
+            if (applyCtx.subledgerEntries().size() == 2)
+            {
+                auto const& real = *applyCtx.subledgerEntries()[0];
+                BEAST_EXPECT(real.getFieldU32(sfFlags) == stfIndeterminate);
+                BEAST_EXPECT(noRngIn(real));
+                // drawn from a different state (2^-32 chance to collide)
+                BEAST_EXPECT(hookCode(real) != visibleCode);
+            }
+            BEAST_EXPECT(applyCtx.view()
+                             .read(keylet::account(alice.id()))
+                             ->isFieldPresent(sfDomain));
+
+            auto const txMeta = applyCtx.apply(tesSUCCESS);
+            BEAST_EXPECT(txMeta && txMeta->hasSubledger());
+            // the real cursor advanced once (bob's one draw), and was
+            // written back on commit
+            BEAST_EXPECT(randomState(ov) == step(seed));
+            BEAST_EXPECT(
+                ov.read(keylet::account(alice.id()))->isFieldPresent(sfDomain));
+            BEAST_EXPECT(
+                ov.read(keylet::account(carol.id()))
+                    ->getFieldAmount(sfBalance) ==
+                closed->read(keylet::account(carol.id()))
+                        ->getFieldAmount(sfBalance) +
+                    STAmount{XRP(100)});
+        }
+
+        {
+            // The owning transaction's own draws use the real cursor while it
+            // has a subledger, and survive a discard.
+            STTx const otxn = makeOtxn();
+            OpenView ov(&*closed);
+            ApplyContext applyCtx = createApplyContext(env, ov, otxn);
+
+            // carol has no hook: no draw, no fork
+            auto const r =
+                applyCtx.applyToSubledger(makePayment(ov, otxn, carol, 1));
+            if (!BEAST_EXPECT(r.entry))
+                return;
+            BEAST_EXPECT(!r.entry->isFieldPresent(sfFlags));
+            BEAST_EXPECT(!applyCtx.subledgerRngDrawn());
+
+            auto hookCtx = makeStubHookContext(
+                applyCtx,
+                alice.id(),
+                alice.id(),
+                {.result = {.isStrong = true}});
+            auto& api = hookCtx.api();
+            auto const roll = api.dice(6);
+            BEAST_EXPECT(roll.has_value() && roll.value() < 6);
+
+            BEAST_EXPECT(applyCtx.subledgerRngDrawn());
+            BEAST_EXPECT(randomState(applyCtx.view()) == seed);
+            BEAST_EXPECT(
+                dynamic_cast<ApplyViewImpl&>(applyCtx.view()).getRandomData() ==
+                seed);
+
+            applyCtx.discard();
+            BEAST_EXPECT(!applyCtx.hasSubledger());
+            BEAST_EXPECT(randomState(applyCtx.view()) == step(seed));
+            BEAST_EXPECT(
+                dynamic_cast<ApplyViewImpl&>(applyCtx.view()).getRandomData() ==
+                seed);
+        }
+
+        {
+            // A shadow draw never touches the Random object or the real
+            // cursor, and records nothing in the inner's metadata.
+            STTx const otxn = makeOtxn();
+            OpenView ov(&*closed);
+            SubledgerRng rng(otxn.getTransactionID(), false);
+            auto const inner = makePayment(ov, otxn, carol, 1);
+
+            {
+                OpenView child(closed_view, ov, &rng, /*shadow=*/true);
+                ApplyContext innerCtx =
+                    createApplyContext(env, child, *inner, tapATOMIC_EMIT);
+                auto hookCtx = makeStubHookContext(
+                    innerCtx,
+                    carol.id(),
+                    alice.id(),
+                    {.result = {.isStrong = true}});
+                auto& api = hookCtx.api();
+                BEAST_EXPECT(api.dice(6).has_value());
+                BEAST_EXPECT(rng.shadowDrawn && !rng.realDrawn);
+                BEAST_EXPECT(rng.real == seed && rng.shadow);
+                BEAST_EXPECT(rng.shadow != seed && rng.shadow != step(seed));
+                BEAST_EXPECT(randomState(innerCtx.view()) == seed);
+                BEAST_EXPECT(!dynamic_cast<ApplyViewImpl&>(innerCtx.view())
+                                  .getRandomData());
+            }
+
+            {
+                OpenView child(closed_view, ov, &rng, /*shadow=*/false);
+                ApplyContext innerCtx =
+                    createApplyContext(env, child, *inner, tapATOMIC_EMIT);
+                auto hookCtx = makeStubHookContext(
+                    innerCtx,
+                    carol.id(),
+                    alice.id(),
+                    {.result = {.isStrong = true}});
+                auto& api = hookCtx.api();
+                BEAST_EXPECT(api.dice(6).has_value());
+                BEAST_EXPECT(rng.realDrawn);
+                BEAST_EXPECT(rng.real == step(seed));
+                BEAST_EXPECT(randomState(innerCtx.view()) == seed);
+                BEAST_EXPECT(!dynamic_cast<ApplyViewImpl&>(innerCtx.view())
+                                  .getRandomData());
+            }
+        }
+    }
+
+    void
+    test_emit_atomic(FeatureBitset features)
+    {
+        testcase("Test emit (atomic rules)");
+        using namespace jtx;
+
+        auto const alice = Account{"alice"};
+
+        using namespace hook_api;
+        Env env{*this, features};
+
+        STTx invokeTx = STTx(ttINVOKE, [&](STObject& obj) {});
+        OpenView ov{*env.current()};
+        ApplyContext applyCtx = createApplyContext(env, ov, invokeTx);
+
+        // atomicBound == true: FirstLedgerSequence == LastLedgerSequence ==
+        // the ledger being built (required by emit_atomic); otherwise the
+        // usual emit() window
+        auto const makeEmitted = [&](TxType tt, bool atomicBound) {
+            return STTx(tt, [&](STObject& obj) {
+                obj[sfAccount] = alice.id();
+                obj[sfSequence] = 0;
+                obj[sfSigningPubKey] = Slice{};
+                obj[sfFirstLedgerSequence] =
+                    atomicBound ? ov.seq() : env.closed()->seq() + 1;
+                obj[sfLastLedgerSequence] =
+                    atomicBound ? ov.seq() : env.closed()->seq() + 5;
+                obj[sfFee] = env.closed()->fees().base;
+
+                auto& emitDetails = obj.peekFieldObject(sfEmitDetails);
+                emitDetails[sfEmitGeneration] = 1;
+                emitDetails[sfEmitBurden] = 1;
+                emitDetails[sfEmitParentTxnID] = invokeTx.getTransactionID();
+                emitDetails[sfEmitNonce] = uint256();
+                emitDetails[sfEmitHookHash] = uint256();
+            });
+        };
+        STTx const emitInvokeTx = makeEmitted(ttINVOKE, false);
+        STTx const atomicInvokeTx = makeEmitted(ttINVOKE, true);
+        // getSerializer() returns by value: keep it alive for the slice
+        Serializer const emitInvokeSer = emitInvokeTx.getSerializer();
+        Serializer const atomicInvokeSer = atomicInvokeTx.getSerializer();
+        auto const blob = emitInvokeSer.slice();          // for emit()
+        auto const atomicBlob = atomicInvokeSer.slice();  // for emit_atomic()
+
+        bool const enabled = env.current()->rules().enabled(featureAtomicEmit);
+
+        {
+            // A1: a weak (default stub) execution may not emit atomically
+            auto hookCtx = makeStubHookContext(
+                applyCtx,
+                alice.id(),
+                alice.id(),
+                {.expected_etxn_count = 1, .nonce_used = {{uint256(0), true}}});
+            auto const result = hookCtx.api().emit(atomicBlob, /*atomic=*/true);
+            BEAST_EXPECT(result.error() == EMISSION_FAILURE);
+        }
+        {
+            // A1: a strong execution may
+            auto hookCtx = makeStubHookContext(
+                applyCtx,
+                alice.id(),
+                alice.id(),
+                {.expected_etxn_count = 1,
+                 .nonce_used = {{uint256(0), true}},
+                 .result = {.isStrong = true}});
+            auto const result = hookCtx.api().emit(atomicBlob, /*atomic=*/true);
+            BEAST_EXPECT(result.has_value());
+
+            // bookkeeping after a successful atomic emission
+            if (result)
+            {
+                hookCtx.api().recordEmission(*result, /*atomic=*/true);
+                BEAST_EXPECT(applyCtx.atomicEmitCount == 1);
+                BEAST_EXPECT(
+                    hookCtx.nonce_consumed.count(uint256(0)) == 1 &&
+                    hookCtx.nonce_consumed[uint256(0)] == true);
+            }
+            applyCtx.atomicEmitCount = 0;
+        }
+        {
+            // A1: a callback execution may not
+            auto hookCtx = makeStubHookContext(
+                applyCtx,
+                alice.id(),
+                alice.id(),
+                {.expected_etxn_count = 1,
+                 .nonce_used = {{uint256(0), true}},
+                 .result = {.isCallback = true, .isStrong = true}});
+            auto const result = hookCtx.api().emit(atomicBlob, /*atomic=*/true);
+            BEAST_EXPECT(result.error() == EMISSION_FAILURE);
+        }
+        {
+            // A2: a hook running inside a subledger (on an emit_atomic txn)
+            // cannot start another one
+            ApplyContext nested =
+                createApplyContext(env, ov, invokeTx, tapATOMIC_EMIT);
+            auto hookCtx = makeStubHookContext(
+                nested,
+                alice.id(),
+                alice.id(),
+                {.expected_etxn_count = 1,
+                 .nonce_used = {{uint256(0), true}},
+                 .result = {.isStrong = true}});
+            BEAST_EXPECT(
+                hookCtx.api().emit(atomicBlob, /*atomic=*/true).error() ==
+                ALREADY_IN_SUBLEDGER);
+            // plain emit() is still fine there
+            BEAST_EXPECT(hookCtx.api().emit(blob).has_value());
+        }
+        {
+            // A4: per transaction cap
+            applyCtx.atomicEmitCount = hook_api::max_atomic_emit;
+            auto hookCtx = makeStubHookContext(
+                applyCtx,
+                alice.id(),
+                alice.id(),
+                {.expected_etxn_count = 1,
+                 .nonce_used = {{uint256(0), true}},
+                 .result = {.isStrong = true}});
+            BEAST_EXPECT(
+                hookCtx.api().emit(atomicBlob, /*atomic=*/true).error() ==
+                EMISSION_FAILURE);
+            // the cap does not apply to plain emit()
+            BEAST_EXPECT(hookCtx.api().emit(blob).has_value());
+            applyCtx.atomicEmitCount = hook_api::max_atomic_emit - 1;
+            BEAST_EXPECT(
+                hookCtx.api().emit(atomicBlob, /*atomic=*/true).has_value());
+            applyCtx.atomicEmitCount = 0;
+        }
+        {
+            // A5: the etxn_reserve budget is shared between the two queues
+            std::string reason;
+            auto tx = std::make_shared<ripple::Transaction>(
+                std::make_shared<ripple::STTx const>(emitInvokeTx),
+                reason,
+                env.app());
+            std::queue<std::shared_ptr<ripple::Transaction>> q;
+            q.push(tx);
+            std::vector<std::shared_ptr<ripple::Transaction>> qv{tx};
+            {
+                auto hookCtx = makeStubHookContext(
+                    applyCtx,
+                    alice.id(),
+                    alice.id(),
+                    {.expected_etxn_count = 1,
+                     .nonce_used = {{uint256(0), true}},
+                     .result = {.emittedTxn = q, .isStrong = true}});
+                BEAST_EXPECT(
+                    hookCtx.api().emit(atomicBlob, /*atomic=*/true).error() ==
+                    TOO_MANY_EMITTED_TXN);
+            }
+            {
+                auto hookCtx = makeStubHookContext(
+                    applyCtx,
+                    alice.id(),
+                    alice.id(),
+                    {.expected_etxn_count = 1,
+                     .nonce_used = {{uint256(0), true}},
+                     .result = {.emittedAtomicTxn = qv, .isStrong = true}});
+                BEAST_EXPECT(
+                    hookCtx.api().emit(blob).error() == TOO_MANY_EMITTED_TXN);
+            }
+        }
+        {
+            // R2: a nonce spent by any emission is refused for emit_atomic,
+            // a nonce spent atomically is refused for emit(); plain emit()
+            // reusing a plain emit() nonce is untouched (legacy)
+            auto mk = [&](bool spentAtomically) {
+                return makeStubHookContext(
+                    applyCtx,
+                    alice.id(),
+                    alice.id(),
+                    {.expected_etxn_count = 1,
+                     .nonce_used = {{uint256(0), true}},
+                     .nonce_consumed = {{uint256(0), spentAtomically}},
+                     .result = {.isStrong = true}});
+            };
+            {
+                auto hookCtx = mk(false);
+                BEAST_EXPECT(
+                    hookCtx.api()
+                        .emit(atomicBlob, /*atomic=*/true)
+                        .has_value() == !enabled);
+                BEAST_EXPECT(hookCtx.api().emit(blob).has_value());
+            }
+            {
+                auto hookCtx = mk(true);
+                BEAST_EXPECT(
+                    hookCtx.api()
+                        .emit(atomicBlob, /*atomic=*/true)
+                        .has_value() == !enabled);
+                BEAST_EXPECT(hookCtx.api().emit(blob).has_value() == !enabled);
+            }
+        }
+        {
+            // ledger window: emit_atomic requires FLS == LLS == current seq,
+            // and such a blob is in turn refused by plain emit()
+            auto hookCtx = makeStubHookContext(
+                applyCtx,
+                alice.id(),
+                alice.id(),
+                {.expected_etxn_count = 1,
+                 .nonce_used = {{uint256(0), true}},
+                 .result = {.isStrong = true}});
+            BEAST_EXPECT(
+                hookCtx.api().emit(blob, /*atomic=*/true).error() ==
+                EMISSION_FAILURE);
+            BEAST_EXPECT(
+                hookCtx.api().emit(atomicBlob).error() == EMISSION_FAILURE);
+            BEAST_EXPECT(
+                hookCtx.api().emit(atomicBlob, /*atomic=*/true).has_value());
+            BEAST_EXPECT(hookCtx.api().emit(blob).has_value());
+        }
+        {
+            // R1: the fee floor is computed against the view being applied
+            // (fix20260929, which emit_atomic relies on: the inner is applied
+            // to that same view)
+            auto hookCtx = makeStubHookContext(
+                applyCtx, alice.id(), alice.id(), {.expected_etxn_count = 1});
+            auto const fee = hookCtx.api().etxn_fee_base(blob);
+            BEAST_EXPECT(fee.has_value());
+            if (fee && env.current()->rules().enabled(fix20260929))
+                BEAST_EXPECT(
+                    *fee ==
+                    ripple::calculateBaseFee(applyCtx.view(), emitInvokeTx)
+                        .drops());
+        }
+        {
+            // tapATOMIC_EMIT defensive checks in preflight1 / checkSign: the
+            // flag on anything that is not an unsigned emitted txn is
+            // refused; on a genuine emitted txn preflight passes without
+            // SF_EMITTED
+            auto const& rules = env.current()->rules();
+            STTx const plain = STTx(ttINVOKE, [&](STObject& obj) {
+                obj[sfAccount] = alice.id();
+                obj[sfSequence] = 1;
+                obj[sfSigningPubKey] = Slice{};
+                obj[sfFee] = env.closed()->fees().base;
+            });
+            BEAST_EXPECT(
+                preflight(env.app(), rules, plain, tapATOMIC_EMIT, env.journal)
+                    .ter == temINVALID);
+
+            STTx signedEmitted = emitInvokeTx;
+            signedEmitted.setFieldVL(
+                sfTxnSignature, std::vector<uint8_t>(64, 1));
+            BEAST_EXPECT(
+                preflight(
+                    env.app(),
+                    rules,
+                    signedEmitted,
+                    tapATOMIC_EMIT,
+                    env.journal)
+                    .ter == temINVALID);
+
+            STTx zeroKey = emitInvokeTx;
+            zeroKey.setFieldVL(sfSigningPubKey, std::vector<uint8_t>(33, 0));
+            BEAST_EXPECT(hook::hasNoSignatureMaterial(zeroKey));
+            BEAST_EXPECT(
+                preflight(
+                    env.app(), rules, zeroKey, tapATOMIC_EMIT, env.journal)
+                    .ter == tesSUCCESS);
+            BEAST_EXPECT(
+                preflight(env.app(), rules, zeroKey, tapNONE, env.journal)
+                    .ter == telNON_LOCAL_EMITTED_TXN);
+        }
     }
 
     void
@@ -5198,6 +6105,12 @@ public:
         test_prepare(features);
         test_emit(features);
         test_emitted_txn_entry(features);
+        // exercises the amendment-off branches test_emit_atomic checks for
+        test_emit_atomic(features - featureAtomicEmit);
+        test_emit_atomic(features);
+        test_emit_atomic_subledger(features);
+        test_emit_atomic_rng(features);
+        test_subledger_merge();
         test_etxn_burden(features);
         test_etxn_generation(features);
         test_otxn_burden(features);

src/test/jtx/hook.h

@@ -55,6 +55,7 @@ struct StubHookResult
     ripple::uint256 const hookNamespace = ripple::uint256();
 
     std::queue<std::shared_ptr<ripple::Transaction>> emittedTxn{};
+    std::vector<std::shared_ptr<ripple::Transaction>> emittedAtomicTxn{};
     std::optional<hook::HookStateMap> stateMap = std::nullopt;
     uint16_t changedStateCount = 0;
     std::map<
@@ -92,6 +93,7 @@ struct StubHookContext
     uint16_t ledger_nonce_counter{0};
     int64_t expected_etxn_count{-1};
     std::map<ripple::uint256, bool> nonce_used{};
+    std::map<ripple::uint256, bool> nonce_consumed{};
     uint32_t generation = 0;
     uint64_t burden = 0;
     std::map<uint32_t, uint32_t> guard_map{};

src/test/jtx/impl/hook.cpp

@@ -117,6 +117,7 @@ makeStubHookContext(
         .ledger_nonce_counter = stubHookContext.ledger_nonce_counter,
         .expected_etxn_count = stubHookContext.expected_etxn_count,
         .nonce_used = stubHookContext.nonce_used,
+        .nonce_consumed = stubHookContext.nonce_consumed,
         .generation = stubHookContext.generation,
         .burden = stubHookContext.burden,
         .guard_map = stubHookContext.guard_map,
@@ -131,6 +132,7 @@ makeStubHookContext(
                 .otxnAccount = otxnAccount,
                 .hookNamespace = result.hookNamespace,
                 .emittedTxn = result.emittedTxn,
+                .emittedAtomicTxn = result.emittedAtomicTxn,
                 .stateMap = stateMap,
                 .changedStateCount = result.changedStateCount,
                 .hookParamOverrides = result.hookParamOverrides,

src/test/ledger/View_test.cpp

@@ -329,6 +329,125 @@ class View_test : public beast::unit_test::suite
         BEAST_EXPECT(!v0.exists(k(4)));
     }
 
+    // OpenView closed_view / subledger_view constructors (emit_atomic) and
+    // ApplyViewBase::flush / rebase
+    void
+    testClosedView()
+    {
+        testcase("Closed nested view");
+
+        using namespace jtx;
+        Env env(*this);
+        auto const open = env.current();
+
+        auto const txn = std::make_shared<Serializer>();
+        txn->add32(1);
+
+        // existing constructors are unchanged: txCount() counts only the
+        // view's own transactions and txExists() does not consult the base
+        OpenView base(&*open);
+        BEAST_EXPECT(base.open());
+        BEAST_EXPECT(base.txCount() == 0);
+        base.rawTxInsert(uint256(1), txn, nullptr);
+        BEAST_EXPECT(base.txCount() == 1);
+        BEAST_EXPECT(base.txExists(uint256(1)));
+        {
+            OpenView plain(&base);
+            BEAST_EXPECT(plain.txCount() == 0);
+            BEAST_EXPECT(!plain.txExists(uint256(1)));
+            OpenView copy(base);
+            BEAST_EXPECT(copy.txCount() == 1);
+        }
+
+        // closed view: always closed, same sequence, contiguous txCount,
+        // txExists delegated to the base (one level)
+        OpenView sandbox(closed_view, base);
+        BEAST_EXPECT(!sandbox.open());
+        BEAST_EXPECT(sandbox.seq() == base.seq());
+        BEAST_EXPECT(sandbox.info().parentHash == base.info().parentHash);
+        BEAST_EXPECT(sandbox.txCount() == 1);
+        BEAST_EXPECT(sandbox.txExists(uint256(1)));
+        BEAST_EXPECT(!sandbox.txExists(uint256(2)));
+        sandbox.rawTxInsert(uint256(2), txn, txn);
+        BEAST_EXPECT(sandbox.txCount() == 2);
+        BEAST_EXPECT(base.txCount() == 1);
+        {
+            OpenView copy(sandbox);
+            BEAST_EXPECT(!copy.open());
+            BEAST_EXPECT(copy.txCount() == 2);
+            BEAST_EXPECT(copy.txExists(uint256(1)));
+        }
+
+        // applyState propagates state (and destroyed drops) but no txns
+        BEAST_EXPECT(sandbox.stateSize() == 0);
+        sandbox.rawInsert(sle(7, 7));
+        sandbox.rawDestroyXRP(XRPAmount{5});
+        BEAST_EXPECT(sandbox.stateSize() == 1);
+        sandbox.applyState(base);
+        BEAST_EXPECT(base.exists(k(7)));
+        BEAST_EXPECT(!base.txExists(uint256(2)));
+        BEAST_EXPECT(base.txCount() == 1);
+
+        // apply propagates both
+        OpenView sandbox2(closed_view, base);
+        sandbox2.rawTxInsert(uint256(3), txn, txn);
+        sandbox2.rawInsert(sle(8, 8));
+        sandbox2.apply(base);
+        BEAST_EXPECT(base.exists(k(8)));
+        BEAST_EXPECT(base.txExists(uint256(3)));
+        BEAST_EXPECT(base.txCount() == 2);
+
+        // subledger view: open() follows the base, its own transactions are
+        // numbered from zero, txExists() is delegated; a closed_view child
+        // of it continues its numbering and applies into it
+        {
+            OpenView sub(subledger_view, base);
+            BEAST_EXPECT(sub.open() == base.open());
+            BEAST_EXPECT(sub.seq() == base.seq());
+            BEAST_EXPECT(sub.txCount() == 0);
+            BEAST_EXPECT(sub.txExists(uint256(1)));
+            BEAST_EXPECT(sub.exists(k(8)));
+
+            OpenView child(closed_view, sub);
+            BEAST_EXPECT(!child.open());
+            BEAST_EXPECT(child.txCount() == 0);
+            child.rawTxInsert(uint256(9), txn, txn);
+            child.rawInsert(sle(9, 9));
+            child.apply(sub);
+            BEAST_EXPECT(sub.txCount() == 1);
+            BEAST_EXPECT(sub.exists(k(9)));
+
+            OpenView child2(closed_view, sub);
+            BEAST_EXPECT(child2.txCount() == 1);
+            BEAST_EXPECT(child2.txExists(uint256(9)));
+            BEAST_EXPECT(child2.txExists(uint256(1)));
+
+            BEAST_EXPECT(!base.txExists(uint256(9)));
+            BEAST_EXPECT(!base.exists(k(9)));
+        }
+
+        // flush moves the buffered changes down, rebase re-layers the
+        // (empty) view on top of the view they went to
+        {
+            OpenView lower(&base);
+            ApplyViewImpl av(&base, tapNONE);
+            av.insert(sle(10, 10));
+            BEAST_EXPECT(av.size() == 1);
+            av.flush(lower);
+            BEAST_EXPECT(av.size() == 0);
+            BEAST_EXPECT(lower.exists(k(10)));
+            BEAST_EXPECT(!base.exists(k(10)));
+            av.rebase(&lower);
+            BEAST_EXPECT(av.exists(k(10)));
+            auto const e = av.peek(k(10));
+            BEAST_EXPECT(e != nullptr);
+            if (e)
+                av.erase(e);
+            BEAST_EXPECT(!av.exists(k(10)));
+            BEAST_EXPECT(lower.exists(k(10)));
+        }
+    }
+
     // Verify contextual information
     void
     testContext()
@@ -1098,6 +1217,7 @@ class View_test : public beast::unit_test::suite
         testMeta();
         testMetaSucc();
         testStacked();
+        testClosedView();
         testContext();
         testSles();
         testUpperAndLowerBound();

src/xrpld/app/hook/HookAPI.h

@@ -345,11 +345,34 @@ public:
     // sto_erase(): same as sto_emplace with field_object = nullopt
 
     /// etxn APIs
+    // atomic == true (prepare_atomic): the ledger window is set to exactly
+    // the ledger being built (FirstLedgerSequence == LastLedgerSequence ==
+    // current seq), as emit_atomic requires; otherwise the emit() window.
     Expected<Bytes, HookReturnCode>
-    prepare(Slice const& txBlob) const;
+    prepare(Slice const& txBlob, bool atomic = false) const;
 
+    // atomic == true applies the emit_atomic rules (strong only, not inside
+    // a subledger, per-transaction cap) on top of the emit rules. Validation
+    // only: nothing is emitted or executed.
     Expected<std::shared_ptr<Transaction>, HookReturnCode>
-    emit(Slice const& txBlob) const;
+    emit(Slice const& txBlob, bool atomic = false) const;
+
+    // Bookkeeping for an emission: marks the nonce as consumed and, for
+    // atomic emissions, bumps the per-transaction attempt counter on the
+    // ApplyContext. Plain emit() callers must call this after queueing the
+    // txn; apply_atomic() calls it itself (before executing, so an attempt
+    // that fails still counts).
+    void
+    recordEmission(std::shared_ptr<Transaction> const& tpTrans, bool atomic)
+        const;
+
+    // emit_atomic: execute a txn accepted by emit(txBlob, true) in this
+    // transaction's subledger. The emission is recorded (recordEmission)
+    // first, whatever the outcome. On success the txn was applied (tes or tec)
+    // and its SubledgerTransaction entry is returned. Otherwise
+    // EMISSION_FAILURE, and the subledger holds nothing from the txn.
+    Expected<std::shared_ptr<STObject const>, HookReturnCode>
+    apply_atomic(std::shared_ptr<Transaction> const& tpTrans) const;
 
     Expected<uint64_t, HookReturnCode>
     etxn_burden() const;
@@ -561,6 +584,15 @@ public:
     Expected<uint32_t, HookReturnCode>
     meta_slot(uint32_t slot_into) const;
 
+    // Slot one SubledgerTransaction { EmittedTxnID, EmittedTxn,
+    // TransactionMetaData } of this transaction's subledger: the one with
+    // txid if given, else the one at index (application order).
+    Expected<uint32_t, HookReturnCode>
+    subledger_slot(
+        uint32_t slot_into,
+        std::optional<uint256> const& txid,
+        uint32_t index) const;
+
     Expected<std::pair<uint32_t, uint32_t>, HookReturnCode>
     xpop_slot(uint32_t slot_into_tx, uint32_t slot_into_meta) const;
 

src/xrpld/app/hook/applyHook.h

@@ -88,6 +88,13 @@ canHook(
     ripple::uint256 hookOn,
     std::optional<ripple::Blob> hookName);
 
+// True iff the (emitted) txn carries no signature material: no
+// sfTxnSignature, no sfSigners, and an sfSigningPubKey that is either
+// empty or 33 zero bytes. Mirrors HookAPI::emit rules 2/2.a/4; used by
+// Transactor::checkSign for tapATOMIC_EMIT inners. Keep in sync.
+bool
+hasNoSignatureMaterial(ripple::STTx const& tx);
+
 bool
 canEmit(ripple::TxType txType, ripple::uint256 hookCanEmit);
 
@@ -155,6 +162,9 @@ struct HookResult
 
     std::queue<std::shared_ptr<ripple::Transaction>>
         emittedTxn{};  // etx stored here until accept/rollback
+    std::vector<std::shared_ptr<ripple::Transaction>>
+        emittedAtomicTxn{};  // emit_atomic txns this execution put in the
+                             // subledger (they share the etxn_reserve budget)
     HookStateMap& stateMap;
     uint16_t changedStateCount = 0;
     std::map<
@@ -215,6 +225,10 @@ struct HookContext
     int64_t expected_etxn_count{-1};  // make this a 64bit int so the uint32
                                       // from the hookapi cant overflow it
     std::map<ripple::uint256, bool> nonce_used{};
+    // nonces already spent by an emission in this execution (by any
+    // emit_atomic attempt, or by a successful emit):
+    // nonce -> true iff spent by emit_atomic (featureAtomicEmit)
+    std::map<ripple::uint256, bool> nonce_consumed{};
     uint32_t generation =
         0;  // used for caching, only generated when txn_generation is called
     uint64_t burden =

src/xrpld/app/hook/detail/HookAPI.cpp

@@ -418,7 +418,7 @@ HookAPI::sto_emplace(
 
 /// etxn APIs
 Expected<Bytes, HookReturnCode>
-HookAPI::prepare(Slice const& txBlob) const
+HookAPI::prepare(Slice const& txBlob, bool atomic) const
 {
     auto& applyCtx = hookCtx.applyCtx;
     auto j = applyCtx.app.journal("View");
@@ -458,11 +458,21 @@ HookAPI::prepare(Slice const& txBlob) const
     json[jss::Account] = raddr;
 
     uint32_t seq = applyCtx.view().info().seq;
-    if (!json.isMember(jss::FirstLedgerSequence))
-        json[jss::FirstLedgerSequence] = Json::Value(seq + 1);
+    if (atomic)
+    {
+        // emit_atomic only accepts a window of exactly this ledger, so set
+        // both bounds regardless of what the caller put in the blob
+        json[jss::FirstLedgerSequence] = Json::Value(seq);
+        json[jss::LastLedgerSequence] = Json::Value(seq);
+    }
+    else
+    {
+        if (!json.isMember(jss::FirstLedgerSequence))
+            json[jss::FirstLedgerSequence] = Json::Value(seq + 1);
 
-    if (!json.isMember(jss::LastLedgerSequence))
-        json[jss::LastLedgerSequence] = Json::Value(seq + 5);
+        if (!json.isMember(jss::LastLedgerSequence))
+            json[jss::LastLedgerSequence] = Json::Value(seq + 5);
+    }
 
     uint8_t details[512];
     if (!json.isMember(jss::EmitDetails))
@@ -534,7 +544,7 @@ HookAPI::prepare(Slice const& txBlob) const
 }
 
 Expected<std::shared_ptr<Transaction>, HookReturnCode>
-HookAPI::emit(Slice const& txBlob) const
+HookAPI::emit(Slice const& txBlob, bool atomic) const
 {
     auto& applyCtx = hookCtx.applyCtx;
     auto j = applyCtx.app.journal("View");
@@ -543,9 +553,49 @@ HookAPI::emit(Slice const& txBlob) const
     if (hookCtx.expected_etxn_count < 0)
         return Unexpected(PREREQUISITE_NOT_MET);
 
-    if (hookCtx.result.emittedTxn.size() >= hookCtx.expected_etxn_count)
+    // emit and emit_atomic share the etxn_reserve() budget; the atomic
+    // queue is always empty unless featureAtomicEmit is enabled.
+    if (hookCtx.result.emittedTxn.size() +
+            hookCtx.result.emittedAtomicTxn.size() >=
+        hookCtx.expected_etxn_count)
         return Unexpected(TOO_MANY_EMITTED_TXN);
 
+    if (atomic)
+    {
+        // Only a strong execution can roll the parent back, so only a
+        // strong execution may make the parent depend on the inner txn.
+        // (weak TSH, hook_again re-execution and cbak all run isStrong=false)
+        if (!hookCtx.result.isStrong || hookCtx.result.isCallback)
+        {
+            JLOG(j.trace()) << "HookEmit[" << HC_ACC()
+                            << "]: emit_atomic is only allowed from a strong "
+                               "(pre-application) hook execution.";
+            return Unexpected(EMISSION_FAILURE);
+        }
+
+        // A subledger's txns cannot have subledgers of their own. Every
+        // hook that runs while an atomic inner txn is being applied sees
+        // tapATOMIC_EMIT through the inner's ApplyContext.
+        if (applyCtx.flags() & tapATOMIC_EMIT)
+        {
+            JLOG(j.trace()) << "HookEmit[" << HC_ACC()
+                            << "]: emit_atomic called inside a subledger.";
+            return Unexpected(ALREADY_IN_SUBLEDGER);
+        }
+
+        // Per outer transaction cap, shared across every hook execution
+        if (applyCtx.atomicEmitCount >= hook_api::max_atomic_emit)
+        {
+            // max_atomic_emit is a uint8_t: cast so it logs as a number
+            JLOG(j.trace())
+                << "HookEmit[" << HC_ACC()
+                << "]: too many emit_atomic txns for this "
+                   "transaction (max "
+                << static_cast<unsigned>(hook_api::max_atomic_emit) << ").";
+            return Unexpected(EMISSION_FAILURE);
+        }
+    }
+
     std::shared_ptr<STTx const> stpTrans;
     try
     {
@@ -734,6 +784,26 @@ HookAPI::emit(Slice const& txBlob) const
         return Unexpected(EMISSION_FAILURE);
     }
 
+    // A nonce spent by a successful emission cannot be reused when
+    // emit_atomic is involved on either side. (Two identical blobs would
+    // share a txid: two atomic copies would collide inside the sandbox, and
+    // an atomic + a normal copy would land in two ledgers.) Plain emit()
+    // reusing a nonce spent by plain emit() is left as-is: finalizeHookResult
+    // silently de-duplicates it, and changing that could break deployed
+    // hooks.
+    if (view.rules().enabled(featureAtomicEmit))
+    {
+        auto const consumed = hookCtx.nonce_consumed.find(nonce);
+        if (consumed != hookCtx.nonce_consumed.end() &&
+            (atomic || consumed->second))
+        {
+            JLOG(j.trace()) << "HookEmit[" << HC_ACC()
+                            << "]: sfEmitNonce was already used by a previous "
+                               "emission in this execution";
+            return Unexpected(EMISSION_FAILURE);
+        }
+    }
+
     if (callback && *callback != hookCtx.result.account)
     {
         JLOG(j.trace()) << "HookEmit[" << HC_ACC()
@@ -768,30 +838,50 @@ HookAPI::emit(Slice const& txBlob) const
 
     uint32_t tx_lls = stpTrans->getFieldU32(sfLastLedgerSequence);
     uint32_t ledgerSeq = view.info().seq;
-    if (tx_lls < ledgerSeq + 1)
+    if (atomic)
     {
-        JLOG(j.trace())
-            << "HookEmit[" << HC_ACC()
-            << "]: sfLastLedgerSequence invalid (less than next ledger)";
-        return Unexpected(EMISSION_FAILURE);
+        // An atomic inner txn only ever lands in the ledger currently being
+        // built, so its validity window is exactly that ledger:
+        // FirstLedgerSequence == LastLedgerSequence == current seq. (This is
+        // stricter than prepare()'s defaults; hooks set both fields in the
+        // blob before calling prepare(), which keeps fields already present.)
+        if (tx_lls != ledgerSeq ||
+            !stpTrans->isFieldPresent(sfFirstLedgerSequence) ||
+            stpTrans->getFieldU32(sfFirstLedgerSequence) != ledgerSeq)
+        {
+            JLOG(j.trace()) << "HookEmit[" << HC_ACC()
+                            << "]: emit_atomic requires FirstLedgerSequence "
+                               "== LastLedgerSequence == current ledger seq";
+            return Unexpected(EMISSION_FAILURE);
+        }
     }
-
-    if (tx_lls > ledgerSeq + 5)
+    else
     {
-        JLOG(j.trace())
-            << "HookEmit[" << HC_ACC()
-            << "]: sfLastLedgerSequence cannot be greater than current seq + 5";
-        return Unexpected(EMISSION_FAILURE);
-    }
+        if (tx_lls < ledgerSeq + 1)
+        {
+            JLOG(j.trace())
+                << "HookEmit[" << HC_ACC()
+                << "]: sfLastLedgerSequence invalid (less than next ledger)";
+            return Unexpected(EMISSION_FAILURE);
+        }
 
-    // rule 6
-    if (!stpTrans->isFieldPresent(sfFirstLedgerSequence) ||
-        stpTrans->getFieldU32(sfFirstLedgerSequence) > tx_lls)
-    {
-        JLOG(j.trace()) << "HookEmit[" << HC_ACC()
-                        << "]: sfFirstLedgerSequence must be present and <= "
-                           "LastLedgerSequence";
-        return Unexpected(EMISSION_FAILURE);
+        if (tx_lls > ledgerSeq + 5)
+        {
+            JLOG(j.trace()) << "HookEmit[" << HC_ACC()
+                            << "]: sfLastLedgerSequence cannot be greater "
+                               "than current seq + 5";
+            return Unexpected(EMISSION_FAILURE);
+        }
+
+        // rule 6
+        if (!stpTrans->isFieldPresent(sfFirstLedgerSequence) ||
+            stpTrans->getFieldU32(sfFirstLedgerSequence) > tx_lls)
+        {
+            JLOG(j.trace()) << "HookEmit[" << HC_ACC()
+                            << "]: sfFirstLedgerSequence must be present and "
+                               "<= LastLedgerSequence";
+            return Unexpected(EMISSION_FAILURE);
+        }
     }
 
     // rule 7 check the emitted txn pays the appropriate fee
@@ -848,6 +938,50 @@ HookAPI::emit(Slice const& txBlob) const
     return tpTrans;
 }
 
+void
+HookAPI::recordEmission(
+    std::shared_ptr<Transaction> const& tpTrans,
+    bool atomic) const
+{
+    auto const& stx = *tpTrans->getSTransaction();
+    auto const& emitDetails = const_cast<ripple::STTx&>(stx)
+                                  .getField(sfEmitDetails)
+                                  .downcast<STObject>();
+    auto const nonce = emitDetails.getFieldH256(sfEmitNonce);
+    // once a nonce has been spent atomically it stays marked atomic
+    hookCtx.nonce_consumed[nonce] |= atomic;
+    if (atomic)
+        ++hookCtx.applyCtx.atomicEmitCount;
+}
+
+Expected<std::shared_ptr<STObject const>, HookReturnCode>
+HookAPI::apply_atomic(std::shared_ptr<Transaction> const& tpTrans) const
+{
+    auto& applyCtx = hookCtx.applyCtx;
+    auto j = applyCtx.app.journal("View");
+
+    // The attempt is spent whatever its outcome (it may already have run the
+    // inner txn's hooks): consume the nonce and bump the per-transaction
+    // emit_atomic counter before executing. Doing this here, rather than
+    // leaving it to the caller, means no path can apply an atomic emission
+    // without also making its nonce unusable for a second emission.
+    recordEmission(tpTrans, /*atomic=*/true);
+
+    auto const r = applyCtx.applyToSubledger(tpTrans->getSTransaction());
+    if (!r.entry)
+    {
+        JLOG(j.trace()) << "HookEmit[" << HC_ACC() << "]: emit_atomic txn "
+                        << tpTrans->getID() << " did not enter the subledger: "
+                        << transToken(r.ter);
+        return Unexpected(EMISSION_FAILURE);
+    }
+
+    JLOG(j.trace()) << "HookEmit[" << HC_ACC() << "]: emit_atomic txn "
+                    << tpTrans->getID()
+                    << " applied in the subledger: " << transToken(r.ter);
+    return r.entry;
+}
+
 Expected<uint64_t, HookReturnCode>
 HookAPI::etxn_burden() const
 {
@@ -1932,15 +2066,29 @@ HookAPI::ledger_nonce() const
 //
 // The state the transaction's first draw saw is written to the transaction's
 // final metadata (sfRandomData), never its provisional metadata.
+//
+// featureAtomicEmit: while a transaction has a subledger, draws do not touch
+// the Random object. They iterate one of its SubledgerRng cursors instead:
+// the real one (the transaction's own Hooks, and the real application of an
+// inner transaction), or the shadow one (the application of an inner
+// transaction to the subledger the strong Hooks see), which is forked off the
+// real one with H(shadow, real state, txid of the owning transaction). The
+// real cursor is written back to the Random object when the owning
+// transaction is committed or discarded. Inner transactions never record
+// sfRandomData.
 Expected<uint256, HookReturnCode>
 HookAPI::rng_next() const
 {
     static constexpr std::uint32_t rngOutputTag = 0x524E474F;  // 'RNGO'
     static constexpr std::uint32_t rngStepTag = 0x524E4749;    // 'RNGI'
+    static constexpr std::uint32_t rngShadowTag = 0x524E4753;  // 'RNGS'
 
     auto& applyCtx = hookCtx.applyCtx;
     auto& view = applyCtx.view();
 
+    bool shadow = false;
+    SubledgerRng* const srng = applyCtx.subledgerRng(shadow);
+
     if (!view.rules().enabled(featureRNG))
         return Unexpected(TOO_LITTLE_ENTROPY);
 
@@ -1954,9 +2102,12 @@ HookAPI::rng_next() const
     // A closed ledger may only draw on its own seed. Open ledger application
     // is a preview, so it may run on the last closed ledger's state; the
     // final result is decided when the ledger closes.
+    // (an inner transaction's view is always closed: use the openness of the
+    // transaction that owns the subledger)
     auto const seq = view.seq();
     auto const seededFor = sle->getFieldU32(sfLedgerSequence);
-    if (seededFor != seq && !(view.open() && seededFor + 1 == seq))
+    bool const open = srng ? srng->open : view.open();
+    if (seededFor != seq && !(open && seededFor + 1 == seq))
         return Unexpected(TOO_LITTLE_ENTROPY);
 
     // At least 80% of the UNLReport must have contributed (at least one
@@ -1967,10 +2118,35 @@ HookAPI::rng_next() const
     if (count == 0 || count * 5 < denominator * 4)
         return Unexpected(TOO_LITTLE_ENTROPY);
 
-    uint256 const state = sle->getFieldH256(sfRandomData);
+    uint256 state;
+    if (srng)
+    {
+        if (!srng->real)
+            srng->real = sle->getFieldH256(sfRandomData);
+        if (shadow && !srng->shadow)
+            srng->shadow =
+                ripple::sha512Half(rngShadowTag, *srng->real, srng->otxnID);
 
-    if (auto* avi = dynamic_cast<ApplyViewImpl*>(&view))
-        avi->setRandomData(state);
+        auto& cursor = shadow ? srng->shadow : srng->real;
+        state = *cursor;
+        cursor = ripple::sha512Half(rngStepTag, state);
+        (shadow ? srng->shadowDrawn : srng->realDrawn) = true;
+
+        // only the transaction that owns the subledger records its first
+        // draw
+        if (!(applyCtx.flags() & tapATOMIC_EMIT))
+        {
+            if (auto* avi = dynamic_cast<ApplyViewImpl*>(&view))
+                avi->setRandomData(state);
+        }
+    }
+    else
+    {
+        state = sle->getFieldH256(sfRandomData);
+
+        if (auto* avi = dynamic_cast<ApplyViewImpl*>(&view))
+            avi->setRandomData(state);
+    }
 
     auto const& hr = hookCtx.result;
     uint256 const out = ripple::sha512Half(
@@ -1984,8 +2160,11 @@ HookAPI::rng_next() const
         static_cast<std::uint8_t>(hr.isCallback ? 1 : 0),
         hr.wasmParam);
 
-    sle->setFieldH256(sfRandomData, ripple::sha512Half(rngStepTag, state));
-    view.update(sle);
+    if (!srng)
+    {
+        sle->setFieldH256(sfRandomData, ripple::sha512Half(rngStepTag, state));
+        view.update(sle);
+    }
 
     ++hookCtx.rng_draws;
     return out;
@@ -2625,6 +2804,54 @@ HookAPI::meta_slot(uint32_t slot_into) const
     return slot_into;
 }
 
+Expected<uint32_t, HookReturnCode>
+HookAPI::subledger_slot(
+    uint32_t slot_into,
+    std::optional<uint256> const& txid,
+    uint32_t index) const
+{
+    if (slot_into > hook_api::max_slots)
+        return Unexpected(INVALID_ARGUMENT);
+
+    auto const& entries = hookCtx.applyCtx.subledgerEntries();
+
+    std::shared_ptr<STObject const> found;
+    if (txid)
+    {
+        for (auto const& entry : entries)
+        {
+            if (entry->getFieldH256(sfEmittedTxnID) == *txid)
+            {
+                found = entry;
+                break;
+            }
+        }
+    }
+    else if (index < entries.size())
+        found = entries[index];
+
+    if (!found)
+        return Unexpected(DOESNT_EXIST);
+
+    // check if we can emplace the object to a slot
+    if (slot_into == 0 && no_free_slots())
+        return Unexpected(NO_FREE_SLOTS);
+
+    if (slot_into == 0)
+    {
+        if (auto free = get_free_slot(); free)
+            slot_into = *free;
+        else
+            return Unexpected(NO_FREE_SLOTS);
+    }
+
+    hookCtx.slot[slot_into] = hook::SlotEntry{.storage = found, .entry = 0};
+
+    hookCtx.slot[slot_into].entry = &(*hookCtx.slot[slot_into].storage);
+
+    return slot_into;
+}
+
 Expected<std::pair<uint32_t, uint32_t>, HookReturnCode>
 HookAPI::xpop_slot(uint32_t slot_into_tx, uint32_t slot_into_meta) const
 {

src/xrpld/app/hook/detail/applyHook.cpp

@@ -16,6 +16,7 @@
 #include <xrpl/protocol/st.h>
 #include <xrpl/protocol/tokens.h>
 #include <boost/multiprecision/cpp_dec_float.hpp>
+#include <algorithm>
 #include <memory>
 #include <optional>
 #include <string>
@@ -672,6 +673,22 @@ hook::isEmittedTxn(ripple::STTx const& tx)
     return tx.isFieldPresent(ripple::sfEmitDetails);
 }
 
+bool
+hook::hasNoSignatureMaterial(ripple::STTx const& tx)
+{
+    if (tx.isFieldPresent(sfTxnSignature) || tx.isFieldPresent(sfSigners))
+        return false;
+    if (!tx.isFieldPresent(sfSigningPubKey))
+        return false;
+    auto const pk = tx.getSigningPubKey();
+    // prepare()/etxn_details() produce a 33 byte all-zero key; an empty key
+    // is also accepted (HookAPI::emit rule 2)
+    if (pk.size() != 0 && pk.size() != 33)
+        return false;
+    return std::all_of(
+        pk.begin(), pk.end(), [](std::uint8_t b) { return b == 0; });
+}
+
 int64_t
 hook::computeExecutionFee(uint64_t instructionCount)
 {
@@ -1597,6 +1614,11 @@ hook::finalizeHookResult(
                 }
             }
         }
+
+        // emit_atomic txns were already executed in the subledger, which the
+        // transaction's metadata records (Subledger). They are not written
+        // to the emitted directory and not listed in HookEmissions, which
+        // only lists txns that appear in later ledgers.
     }
 
     // add a metadata entry for this hook execution result
@@ -1620,9 +1642,17 @@ hook::finalizeHookResult(
             ripple::Slice{
                 hookResult.exitReason.data(), hookResult.exitReason.size()});
         meta.setFieldU64(sfHookInstructionCount, hookResult.instructionCount);
+        // emit_atomic txns are emissions too (they share etxn_reserve), but
+        // they only exist if the transaction commits its subledger, which
+        // doEmit is the gate for. They are never in HookEmissions (that
+        // array lists txns that will appear in later ledgers); see the
+        // Subledger array of the transaction's metadata instead.
+        std::size_t const emitCount = emission_txnid.size() +
+            (doEmit ? hookResult.emittedAtomicTxn.size() : 0);
         meta.setFieldU16(
             sfHookEmitCount,
-            emission_txnid.size());  // this will never wrap, hard limit
+            static_cast<std::uint16_t>(
+                emitCount));  // this will never wrap, hard limit
         meta.setFieldU16(sfHookExecutionIndex, exec_index);
         meta.setFieldU16(sfHookStateChangeCount, hookResult.changedStateCount);
         meta.setFieldH256(sfHookHash, hookResult.hookHash);
@@ -2707,12 +2737,11 @@ DEFINE_HOOK_FUNCTION(
     HOOK_TEARDOWN();
 }
 
-/* Emit a transaction from this hook. Transaction must be in STObject form,
- * fully formed and valid. XRPLD does not modify transactions it only checks
- * them for validity. */
+/* Same as prepare() but for emit_atomic(): FirstLedgerSequence and
+ * LastLedgerSequence are set to the current ledger sequence. */
 DEFINE_HOOK_FUNCTION(
     int64_t,
-    emit,
+    prepare_atomic,
     uint32_t write_ptr,
     uint32_t write_len,
     uint32_t read_ptr,
@@ -2721,6 +2750,57 @@ DEFINE_HOOK_FUNCTION(
     HOOK_SETUP();  // populates memory_ctx, memory, memory_length, applyCtx,
                    // hookCtx on current stack
 
+    // see emit_atomic: the host function is registered regardless of the
+    // amendment, gate at runtime
+    if (!applyCtx.view().rules().enabled(featureAtomicEmit))
+        return NOT_IMPLEMENTED;  // LCOV_EXCL_LINE
+
+    if (NOT_IN_BOUNDS(read_ptr, read_len, memory_length))
+        return OUT_OF_BOUNDS;
+
+    if (NOT_IN_BOUNDS(write_ptr, write_len, memory_length))
+        return OUT_OF_BOUNDS;
+
+    ripple::Slice txBlob{
+        reinterpret_cast<const void*>(memory + read_ptr), read_len};
+
+    auto const res = api.prepare(txBlob, /*atomic=*/true);
+    if (!res)
+        return res.error();
+
+    auto tx_blob = res.value();
+
+    WRITE_WASM_MEMORY_AND_RETURN(
+        write_ptr,
+        tx_blob.size(),
+        tx_blob.data(),
+        tx_blob.size(),
+        memory,
+        memory_length);
+
+    HOOK_TEARDOWN();
+}
+
+// Shared body of emit()/emit_atomic(): the two host functions differ only in
+// the amendment gate, the `atomic` argument and which queue the emitted txn
+// lands in. `memoryCtx` is needed (not just `memory`/`memory_length`)
+// because WRITE_WASM_MEMORY_AND_RETURN writes through it directly.
+inline std::variant<uint64_t, hook_api::hook_return_code>
+emit_txn(
+    hook::HookContext& hookCtx,
+    WasmEdge_MemoryInstanceContext* memoryCtx,
+    unsigned char* memory,
+    uint64_t memory_length,
+    uint32_t write_ptr,
+    uint32_t write_len,
+    uint32_t read_ptr,
+    uint32_t read_len,
+    bool atomic)
+{
+    using enum hook_api::hook_return_code;
+    auto j = hookCtx.applyCtx.app.journal("View");
+    auto& api = hookCtx.api();
+
     if (NOT_IN_BOUNDS(read_ptr, read_len, memory_length))
         return OUT_OF_BOUNDS;
 
@@ -2730,11 +2810,10 @@ DEFINE_HOOK_FUNCTION(
     if (write_len < 32)
         return TOO_SMALL;
 
-    // Delegate to decoupled HookAPI for emit logic
     ripple::Slice txBlob{
         reinterpret_cast<const void*>(memory + read_ptr), read_len};
 
-    auto const res = api.emit(txBlob);
+    auto const res = api.emit(txBlob, atomic);
 
     if (!res)
         return res.error();
@@ -2748,6 +2827,16 @@ DEFINE_HOOK_FUNCTION(
     if (NOT_IN_BOUNDS(write_ptr, txID.size(), memory_length))
         return OUT_OF_BOUNDS;
 
+    if (atomic)
+    {
+        // Execute it now, in the subledger. apply_atomic() records the
+        // emission first: the attempt (which may run the inner txn's hooks)
+        // is spent whatever its outcome.
+        if (auto const applied = api.apply_atomic(tpTrans); !applied)
+            return applied.error();
+        hookCtx.result.emittedAtomicTxn.push_back(tpTrans);
+    }
+
     auto const write_txid =
         [&]() -> std::variant<uint64_t, hook_api::hook_return_code> {
         WRITE_WASM_MEMORY_AND_RETURN(
@@ -2764,10 +2853,101 @@ DEFINE_HOOK_FUNCTION(
         return std::get<hook_api::hook_return_code>(result);
 
     auto const value = std::get<uint64_t>(result);
-    if (value == 32)
+    if (value == 32 && !atomic)
+    {
         hookCtx.result.emittedTxn.push(tpTrans);
+        api.recordEmission(tpTrans, false);
+    }
 
     return value;
+}
+
+/* Emit a transaction from this hook. Transaction must be in STObject form,
+ * fully formed and valid. XRPLD does not modify transactions it only checks
+ * them for validity. */
+DEFINE_HOOK_FUNCTION(
+    int64_t,
+    emit,
+    uint32_t write_ptr,
+    uint32_t write_len,
+    uint32_t read_ptr,
+    uint32_t read_len)
+{
+    HOOK_SETUP();  // populates memory_ctx, memory, memory_length, applyCtx,
+                   // hookCtx on current stack
+
+    return emit_txn(
+        hookCtx,
+        memoryCtx,
+        memory,
+        memory_length,
+        write_ptr,
+        write_len,
+        read_ptr,
+        read_len,
+        /*atomic=*/false);
+
+    HOOK_TEARDOWN();
+}
+
+/* Emit a transaction atomically with the transaction this hook is executing
+ * for. The emitted txn is executed immediately, in that transaction's
+ * subledger, before this call returns: it sees (and its effects are seen by)
+ * the ledger as this hook sees it. The subledger is committed as part of the
+ * transaction if the transaction succeeds, and abandoned if a hook rolls back
+ * or the transaction fails in any other way.
+ *
+ * Returns 32 (the txid, written to write_ptr) iff the txn was applied to the
+ * subledger, with tesSUCCESS or with a tec code (fee claimed only). Use
+ * subledger_slot() to inspect its result and metadata, and rollback() if the
+ * outcome is not acceptable. Returns EMISSION_FAILURE if it was invalid or not
+ * applied (tem/tef/tel/ter), and ALREADY_IN_SUBLEDGER when called by a hook
+ * that is itself executing inside a subledger. Same blob format and rules as
+ * emit(), plus: strong execution only, FirstLedgerSequence ==
+ * LastLedgerSequence == the current ledger, at most hook_api::max_atomic_emit
+ * attempts per transaction.
+ *
+ * Hook state: this hook's own state_set() calls are still pending (they are
+ * written once every strong chain has run), so the inner txn's hooks do not
+ * see them, and a key written by both is left as this transaction's hooks
+ * wrote it. Reads through state() do see the inner txn's committed state.
+ *
+ * RNG (featureRNG): if the inner txn's hooks draw from the RNG, what this
+ * hook sees (the return value, state(), subledger_slot()) was drawn from a
+ * shadow RNG and is not what will be committed: the same txn is also applied
+ * with the real RNG to a real subledger the hook cannot see, and that one is
+ * committed. The SubledgerTransaction of such a txn, and of every txn emitted
+ * atomically after it, carries stfIndeterminate in sfFlags. The hook may
+ * rollback(), or continue without knowing the real outcome. If the real
+ * subledger cannot follow what the hook saw (the txn is not applied there,
+ * or a change made by this transaction conflicts with it), the transaction
+ * fails with tecSUBLEDGER_DIVERGED at the end of the strong phase. */
+DEFINE_HOOK_FUNCTION(
+    int64_t,
+    emit_atomic,
+    uint32_t write_ptr,
+    uint32_t write_len,
+    uint32_t read_ptr,
+    uint32_t read_len)
+{
+    HOOK_SETUP();  // populates memory_ctx, memory, memory_length, applyCtx,
+                   // hookCtx on current stack
+
+    // The import whitelist only gates SetHook; the host function itself is
+    // registered regardless of amendment state, so gate at runtime too.
+    if (!applyCtx.view().rules().enabled(featureAtomicEmit))
+        return NOT_IMPLEMENTED;  // LCOV_EXCL_LINE
+
+    return emit_txn(
+        hookCtx,
+        memoryCtx,
+        memory,
+        memory_length,
+        write_ptr,
+        write_len,
+        read_ptr,
+        read_len,
+        /*atomic=*/true);
 
     HOOK_TEARDOWN();
 }
@@ -4032,6 +4212,45 @@ DEFINE_HOOK_FUNCTION(int64_t, meta_slot, uint32_t slot_into)
     HOOK_TEARDOWN();
 }
 
+/* Slot a SubledgerTransaction { EmittedTxnID, EmittedTxn, TransactionMetaData }
+ * from the subledger of the transaction this hook is executing for: the one
+ * whose txid is the 32 bytes at read_ptr, or, when read_len is 0, the one at
+ * index read_ptr (application order, 0 based). */
+DEFINE_HOOK_FUNCTION(
+    int64_t,
+    subledger_slot,
+    uint32_t slot_into,
+    uint32_t read_ptr,
+    uint32_t read_len)
+{
+    HOOK_SETUP();
+
+    // the host function is registered regardless of the amendment
+    if (!applyCtx.view().rules().enabled(featureAtomicEmit))
+        return NOT_IMPLEMENTED;  // LCOV_EXCL_LINE
+
+    std::optional<ripple::uint256> txid;
+    uint32_t index = 0;
+    if (read_len == 32)
+    {
+        if (NOT_IN_BOUNDS(read_ptr, read_len, memory_length))
+            return OUT_OF_BOUNDS;
+        txid = ripple::uint256::fromVoid(memory + read_ptr);
+    }
+    else if (read_len == 0)
+        index = read_ptr;
+    else
+        return INVALID_ARGUMENT;
+
+    auto const result = api.subledger_slot(slot_into, txid, index);
+    if (!result)
+        return result.error();
+
+    return result.value();
+
+    HOOK_TEARDOWN();
+}
+
 DEFINE_HOOK_FUNCTION(
     int64_t,
     xpop_slot,

src/xrpld/app/tx/detail/ApplyContext.cpp

@@ -17,6 +17,7 @@
 */
 //==============================================================================
 
+#include <xrpld/app/tx/apply.h>
 #include <xrpld/app/tx/detail/ApplyContext.h>
 #include <xrpld/app/tx/detail/InvariantCheck.h>
 #include <xrpld/app/tx/detail/Transactor.h>
@@ -25,9 +26,141 @@
 #include <xrpl/json/to_string.h>
 #include <xrpl/protocol/Feature.h>
 #include <xrpl/protocol/Indexes.h>
+#include <xrpl/protocol/TxFlags.h>
+#include <xrpl/protocol/detail/STVar.h>
 
 namespace ripple {
 
+namespace {
+
+// Forwards to another RawView, rewinding the threading fields
+// (PreviousTxnID / PreviousTxnLgrSeq) of every entry to the values the base
+// holds for it, or to "never threaded" for an entry the base does not have.
+//
+// The subledger threads the entries it touches to its own transactions,
+// which never enter the ledger's transaction list. Rewinding lets the owning
+// transaction's metadata record each entry's real previous transaction and
+// thread every touched entry to the owning transaction instead.
+class Unthreader final : public RawView
+{
+public:
+    Unthreader(RawView& to, ReadView const& base, bool forwardDestroyedXRP)
+        : to_(to), base_(base), forwardDestroyedXRP_(forwardDestroyedXRP)
+    {
+    }
+
+    void
+    rawErase(std::shared_ptr<SLE> const& sle) override
+    {
+        to_.rawErase(rewind(sle));
+    }
+
+    void
+    rawInsert(std::shared_ptr<SLE> const& sle) override
+    {
+        to_.rawInsert(rewind(sle));
+    }
+
+    void
+    rawReplace(std::shared_ptr<SLE> const& sle) override
+    {
+        to_.rawReplace(rewind(sle));
+    }
+
+    void
+    rawDestroyXRP(XRPAmount const& fee) override
+    {
+        // an open ledger never destroys fees (see Transactor::finishApply)
+        if (forwardDestroyedXRP_)
+            to_.rawDestroyXRP(fee);
+    }
+
+private:
+    std::shared_ptr<SLE>
+    rewind(std::shared_ptr<SLE> const& sle) const
+    {
+        if (!sle->isFieldPresent(sfPreviousTxnID) &&
+            !sle->isFieldPresent(sfPreviousTxnLgrSeq))
+            return sle;
+
+        auto copy = std::make_shared<SLE>(*sle);
+        auto const orig = base_.read(keylet::unchecked(sle->key()));
+        if (!orig)
+        {
+            // created by the subledger: thread it from scratch
+            if (copy->isFieldPresent(sfPreviousTxnID))
+                copy->setFieldH256(sfPreviousTxnID, uint256{});
+            if (copy->isFieldPresent(sfPreviousTxnLgrSeq))
+                copy->setFieldU32(sfPreviousTxnLgrSeq, 0);
+            return copy;
+        }
+
+        if (orig->isFieldPresent(sfPreviousTxnID))
+            copy->setFieldH256(
+                sfPreviousTxnID, orig->getFieldH256(sfPreviousTxnID));
+        else if (copy->isFieldPresent(sfPreviousTxnID))
+            copy->makeFieldAbsent(sfPreviousTxnID);
+
+        if (orig->isFieldPresent(sfPreviousTxnLgrSeq))
+            copy->setFieldU32(
+                sfPreviousTxnLgrSeq, orig->getFieldU32(sfPreviousTxnLgrSeq));
+        else if (copy->isFieldPresent(sfPreviousTxnLgrSeq))
+            copy->makeFieldAbsent(sfPreviousTxnLgrSeq);
+        return copy;
+    }
+
+    RawView& to_;
+    ReadView const& base_;
+    bool const forwardDestroyedXRP_;
+};
+
+}  // namespace
+
+namespace subledger {
+
+std::shared_ptr<SLE>
+mergeEntry(SLE const& before, SLE const& ours, SLE const& theirs)
+{
+    XRPL_ASSERT(
+        before.key() == ours.key() && before.key() == theirs.key() &&
+            before.getType() == ours.getType() &&
+            before.getType() == theirs.getType(),
+        "ripple::subledger::mergeEntry : same entry");
+
+    auto merged = std::make_shared<SLE>(theirs);
+
+    // Ledger entries are templated: all three hold the same fields (present
+    // or not), so iterating `before` visits every one of them.
+    for (auto const& field : before)
+    {
+        SField const& f = field.getFName();
+        bool const inBefore = before.isFieldPresent(f);
+        bool const inOurs = ours.isFieldPresent(f);
+
+        bool const changed = inOurs != inBefore ||
+            (inOurs && !(ours.peekAtField(f) == before.peekAtField(f)));
+        if (!changed)
+            continue;
+
+        bool const inTheirs = theirs.isFieldPresent(f);
+        if (inTheirs != inBefore ||
+            (inTheirs && !(theirs.peekAtField(f) == before.peekAtField(f))))
+            return nullptr;
+
+        if (inOurs)
+        {
+            detail::STVar copy(ours.peekAtField(f));
+            merged->set(std::move(copy.get()));
+        }
+        else
+            merged->makeFieldAbsent(f);
+    }
+
+    return merged;
+}
+
+}  // namespace subledger
+
 ApplyContext::ApplyContext(
     Application& app_,
     OpenView& base,
@@ -53,7 +186,11 @@ ApplyContext::discard()
     // featureRNG: Hook draws are never undone. If this transaction drew from
     // the RNG, its advanced state survives the discard (as does the record of
     // what its Hooks were given), so a rollback cannot preserve an RNG state
-    // for whatever executes next.
+    // for whatever executes next. Read through the working view before the
+    // subledger (if any) is dropped: a draw made before the first
+    // emit_atomic was flushed into the subledger and is only visible
+    // through it. Draws made while there is a subledger went to the real
+    // cursor (writeBackRng).
     std::optional<uint256> const randomData = view_->getRandomData();
     std::optional<uint256> rngState;
     if (randomData)
@@ -62,7 +199,15 @@ ApplyContext::discard()
             rngState = sle->getFieldH256(sfRandomData);
     }
 
+    // the working view may sit on a subledger: replace it before dropping
+    // the subledgers
     view_.emplace(&base_, flags_);
+    subledger_.reset();
+    subledgerEntries_.clear();
+    realSubledger_.reset();
+    realEntries_.clear();
+    indeterminate_ = false;
+    diverged_ = false;
 
     if (randomData)
         view_->setRandomData(*randomData);
@@ -75,18 +220,355 @@ ApplyContext::discard()
             view_->update(sle);
         }
     }
+
+    writeBackRng();
+    rng_.reset();
 }
 
 std::optional<TxMeta>
 ApplyContext::apply(TER ter)
 {
+    if (subledger_)
+        return commitSubledger(ter);
     return view_->apply(base_, tx, ter, flags_ & tapDRY_RUN, journal);
 }
 
+SubledgerRng*
+ApplyContext::subledgerRng(bool& shadow) const
+{
+    if (flags_ & tapATOMIC_EMIT)
+    {
+        shadow = base_.subledgerRngShadow();
+        return base_.subledgerRng();
+    }
+
+    shadow = false;
+    return rng_.get();
+}
+
+void
+ApplyContext::writeBackRng()
+{
+    if (!rng_ || !rng_->realDrawn || !rng_->real)
+        return;
+
+    if (auto sle = view_->peek(keylet::random()))
+    {
+        sle->setFieldH256(sfRandomData, *rng_->real);
+        view_->update(sle);
+    }
+}
+
+bool
+ApplyContext::stillApplies(ReadView const& after) const
+{
+    auto const id = tx.getAccountID(sfAccount);
+    auto const before = base_.read(keylet::account(id));
+    auto const sle = after.read(keylet::account(id));
+
+    // an account that did not exist (e.g. a first Import) must still not
+    // exist, and one that did must not have been deleted
+    if (!before || !sle)
+        return !before && !sle;
+
+    // e.g. an emitted TicketCreate moves the account's sequence: consuming
+    // this transaction's sequence afterwards would move it back
+    if (!isTesSuccess(Transactor::checkSeqProxy(after, tx, journal)))
+        return false;
+
+    // the fee is charged after the subledger (Transactor::apply)
+    return sle->getFieldAmount(sfBalance).xrp() >=
+        tx.getFieldAmount(sfFee).xrp();
+}
+
+std::shared_ptr<STObject const>
+ApplyContext::makeSubledgerEntry(
+    STTx const& stx,
+    TxMeta const& meta,
+    bool indeterminate) const
+{
+    auto entry = std::make_shared<STObject>(sfSubledgerTransaction);
+    entry->setFieldH256(sfEmittedTxnID, stx.getTransactionID());
+    if (indeterminate)
+        entry->setFieldU32(sfFlags, stfIndeterminate);
+    {
+        Serializer s;
+        stx.add(s);
+        SerialIter sit(s.slice());
+        entry->emplace_back(STObject(sit, sfEmittedTxn));
+    }
+    entry->emplace_back(meta.getAsObject());
+    return entry;
+}
+
+ApplyContext::SubledgerResult
+ApplyContext::applyToSubledger(std::shared_ptr<STTx const> const& stx)
+{
+    XRPL_ASSERT(
+        !(flags_ & tapATOMIC_EMIT),
+        "ripple::ApplyContext::applyToSubledger : not inside a subledger");
+
+    if (!subledger_)
+    {
+        subledger_ = std::make_unique<OpenView>(subledger_view, base_);
+        rng_ =
+            std::make_unique<SubledgerRng>(tx.getTransactionID(), base_.open());
+    }
+
+    // Move the working view's pending changes (e.g. the TouchCount of the
+    // TSH accounts touched so far) into the subledger(s) and continue on top
+    // of the visible one. The working view buffers whole entries, so nothing
+    // may change underneath it once it holds any. The real subledger gets
+    // them merged field by field: they were made on top of the visible one.
+    if (realSubledger_ && !diverged_ && !mergeInto(*realSubledger_))
+    {
+        JLOG(journal.debug()) << "Subledger[" << tx.getTransactionID()
+                              << "]: pending changes conflict with the real "
+                                 "subledger";
+        diverged_ = true;
+    }
+    view_->flush(*subledger_);
+    view_->rebase(subledger_.get());
+
+    auto const id = stx->getTransactionID();
+
+    // The visible application. No tapDRY_RUN: the inner must really apply
+    // to the child (the whole subledger is discarded or committed with this
+    // transaction, which honours tapDRY_RUN). No tapRETRY / tapFAIL_HARD: a
+    // tec inner is final. Its Hooks draw from the shadow RNG.
+    rng_->shadowDrawn = false;
+    OpenView child(closed_view, *subledger_, rng_.get(), /*shadow=*/true);
+    auto const result =
+        ripple::apply(app, child, *stx, tapATOMIC_EMIT, journal);
+
+    if (!result.applied)
+    {
+        JLOG(journal.debug())
+            << "Subledger[" << tx.getTransactionID() << "]: " << id
+            << " not applied: " << transToken(result.ter);
+        return {result.ter, nullptr};
+    }
+
+    if (!stillApplies(child))
+    {
+        JLOG(journal.debug())
+            << "Subledger[" << tx.getTransactionID() << "]: " << id
+            << " refused: the transaction could not be applied after it";
+        return {result.ter, nullptr};
+    }
+
+    // the child is closed, so metadata is always generated
+    if (!result.metadata)
+    {
+        // LCOV_EXCL_START
+        UNREACHABLE(
+            "ripple::ApplyContext::applyToSubledger : missing metadata");
+        return {tefINTERNAL, nullptr};
+        // LCOV_EXCL_STOP
+    }
+
+    // Taint by order only: whether the real outcome turns out to differ must
+    // never reach the Hooks.
+    if (rng_->shadowDrawn)
+        indeterminate_ = true;
+
+    // The real application, needed once the subledgers can differ: this
+    // inner drew (the subledgers fork here, from the subledger as it is
+    // before this inner), or they already have.
+    if (indeterminate_ && !diverged_)
+    {
+        OpenView& realBase = realSubledger_ ? *realSubledger_ : *subledger_;
+        OpenView realChild(closed_view, realBase, rng_.get(), false);
+        auto const real =
+            ripple::apply(app, realChild, *stx, tapATOMIC_EMIT, journal);
+
+        if (!real.applied || !real.metadata || !stillApplies(realChild))
+        {
+            JLOG(journal.debug())
+                << "Subledger[" << tx.getTransactionID() << "]: " << id
+                << " could not be applied to the real subledger: "
+                << transToken(real.ter);
+            diverged_ = true;
+        }
+        else
+        {
+            if (!realSubledger_)
+            {
+                // fork: the subledgers were the same up to this inner (the
+                // copy shares the unchanged entries, which are never modified
+                // in place)
+                realSubledger_ = std::make_unique<OpenView>(*subledger_);
+                realEntries_ = subledgerEntries_;
+            }
+            realChild.apply(*realSubledger_);
+            realEntries_.push_back(
+                makeSubledgerEntry(*stx, *real.metadata, true));
+        }
+    }
+
+    child.apply(*subledger_);
+
+    auto entry = makeSubledgerEntry(*stx, *result.metadata, indeterminate_);
+    subledgerEntries_.push_back(entry);
+    // keep the provisional metadata in step with what is visible
+    syncSubledgerMetaData();
+
+    JLOG(journal.trace()) << "Subledger[" << tx.getTransactionID()
+                          << "]: " << id
+                          << " applied: " << transToken(result.ter)
+                          << (indeterminate_ ? " (indeterminate)" : "");
+    return {result.ter, std::move(entry)};
+}
+
+bool
+ApplyContext::mergeInto(OpenView& real)
+{
+    bool ok = true;
+    view_->visit(
+        *subledger_,
+        [&](uint256 const& key,
+            bool isDelete,
+            std::shared_ptr<SLE const> const& before,
+            std::shared_ptr<SLE const> const& after) {
+            if (!ok)
+                return;
+
+            auto const theirs = real.read(keylet::unchecked(key));
+
+            if (isDelete)
+            {
+                // the entry must be as this transaction saw it
+                if (!before || !theirs || !(*theirs == *before))
+                {
+                    ok = false;
+                    return;
+                }
+                real.rawErase(std::make_shared<SLE>(*theirs));
+                return;
+            }
+
+            if (!before)
+            {
+                // created: the key must still be free
+                if (theirs)
+                {
+                    ok = false;
+                    return;
+                }
+                real.rawInsert(std::make_shared<SLE>(*after));
+                return;
+            }
+
+            if (!theirs)
+            {
+                ok = false;
+                return;
+            }
+
+            auto merged = subledger::mergeEntry(*before, *after, *theirs);
+            if (!merged)
+            {
+                ok = false;
+                return;
+            }
+            real.rawReplace(merged);
+        });
+    return ok;
+}
+
+bool
+ApplyContext::concretizeSubledger()
+{
+    // also called for inner transactions, which never have a subledger
+    if (!subledger_)
+        return true;
+
+    if (diverged_)
+    {
+        JLOG(journal.debug())
+            << "Subledger[" << tx.getTransactionID() << "]: diverged";
+        return false;
+    }
+
+    if (!realSubledger_)
+        return true;
+
+    // Merge into a copy, so that a failure leaves everything in place.
+    auto real = std::make_unique<OpenView>(*realSubledger_);
+    if (!mergeInto(*real) || !stillApplies(*real))
+    {
+        JLOG(journal.debug()) << "Subledger[" << tx.getTransactionID()
+                              << "]: cannot reconcile the real subledger";
+        return false;
+    }
+
+    // The working view's changes are in the real subledger now. Drop them
+    // (into the visible subledger, which goes too) and continue on the real
+    // one.
+    view_->flush(*subledger_);
+    view_->rebase(real.get());
+    subledger_ = std::move(real);
+    subledgerEntries_ = std::move(realEntries_);
+    realSubledger_.reset();
+    realEntries_.clear();
+    syncSubledgerMetaData();
+
+    JLOG(journal.trace()) << "Subledger[" << tx.getTransactionID()
+                          << "]: continuing on the real subledger";
+    return true;
+}
+
+std::optional<TxMeta>
+ApplyContext::commitSubledger(TER ter)
+{
+    // Every path that turns a transaction with a subledger into a tec goes
+    // through discard(), which drops it: reset() on a doApply tec, invariant
+    // failure or oversize, and the dedicated weak-phase oversize block in
+    // Transactor::finishApply.
+    XRPL_ASSERT(
+        isTesSuccess(ter),
+        "ripple::ApplyContext::commitSubledger : transaction succeeded");
+    // concretizeSubledger() ran at the end of the strong phase
+    XRPL_ASSERT(
+        !realSubledger_,
+        "ripple::ApplyContext::commitSubledger : subledger is concrete");
+
+    // Fold the working view into the subledger, then the whole subledger
+    // back into the working view re-layered on the base. The working view
+    // keeps its hook metadata, Subledger metadata (see syncSubledgerMetaData)
+    // and delivered amount, and its AffectedNodes now describe the net change
+    // against the base.
+    view_->flush(*subledger_);
+    view_->rebase(&base_);
+    {
+        Unthreader to(*view_, base_, !base_.open());
+        subledger_->applyState(to);
+    }
+    writeBackRng();
+
+    // the working view no longer refers to it
+    subledger_.reset();
+    subledgerEntries_.clear();
+    rng_.reset();
+    indeterminate_ = false;
+
+    return view_->apply(base_, tx, ter, flags_ & tapDRY_RUN, journal);
+}
+
+void
+ApplyContext::syncSubledgerMetaData()
+{
+    std::vector<STObject> subledger;
+    subledger.reserve(subledgerEntries_.size());
+    for (auto const& entry : subledgerEntries_)
+        subledger.push_back(*entry);
+    view_->setSubledgerMetaData(std::move(subledger));
+}
+
 std::size_t
 ApplyContext::size()
 {
-    return view_->size();
+    return view_->size() + (subledger_ ? subledger_->stateSize() : 0);
 }
 
 void
@@ -96,7 +578,7 @@ ApplyContext::visit(std::function<void(
                         std::shared_ptr<SLE const> const&,
                         std::shared_ptr<SLE const> const&)> const& func)
 {
-    view_->visit(base_, func);
+    view_->visit(viewBase(), func);
 }
 
 TER

src/xrpld/app/tx/detail/ApplyContext.h

@@ -26,11 +26,31 @@
 #include <xrpl/beast/utility/Journal.h>
 #include <xrpl/protocol/STTx.h>
 #include <xrpl/protocol/XRPAmount.h>
+#include <memory>
 #include <optional>
 #include <utility>
+#include <vector>
 
 namespace ripple {
 
+namespace subledger {
+
+/** Three-way merge of one ledger entry, field by field (featureAtomicEmit).
+
+    `before` is the entry as the transaction read it (from the subledger its
+    Hooks see), `ours` is the transaction's version of it and `theirs` is the
+    entry in the real subledger. Every field `ours` changed relative to
+    `before` is taken from `ours`; every other field from `theirs`.
+
+    Returns nullptr on a conflict: a field `ours` changed that `theirs` does
+    not hold as `before` held it. All three must be the same entry (key and
+    type).
+*/
+std::shared_ptr<SLE>
+mergeEntry(SLE const& before, SLE const& ours, SLE const& theirs);
+
+}  // namespace subledger
+
 /** State information when applying a tx. */
 class ApplyContext
 {
@@ -50,6 +70,12 @@ public:
     XRPAmount const baseFee;
     beast::Journal const journal;
 
+    /** Number of emit_atomic attempts made so far for this transaction,
+        across every hook execution. Lives here (not on the ApplyViewImpl)
+        so it survives discard()/reset(); it is never decremented, even when
+        the attempt fails or a hook later rolls back. */
+    std::uint32_t atomicEmitCount = 0;
+
     ApplyView&
     view()
     {
@@ -82,18 +108,131 @@ public:
         view_->deliver(amount);
     }
 
-    /** Discard changes and start fresh. */
+    /** Discard changes and start fresh. The subledger(s), if any, are
+        discarded too. RNG draws made through the subledger cursors are
+        written to the Random object of the fresh view (never undone). */
     void
     discard();
 
-    /** Apply the transaction result to the base. */
+    /** Apply the transaction result to the base.
+
+        If the transaction has a subledger it is committed as part of the
+        transaction: the metadata's AffectedNodes describe the net change
+        (subledger and transaction together, every entry threaded to this
+        transaction) and its Subledger array lists the subledger's
+        transactions with their own metadata. Only this transaction is
+        added to the base's transaction list.
+    */
     std::optional<TxMeta> apply(TER);
 
-    /** Get the number of unapplied changes. */
+    /** Outcome of applyToSubledger(). */
+    struct SubledgerResult
+    {
+        // the inner transaction's result
+        TER ter;
+        // its SubledgerTransaction entry; set iff the inner transaction was
+        // applied (tes or tec) and merged into the subledger
+        std::shared_ptr<STObject const> entry;
+    };
+
+    /** Apply an atomically emitted (emit_atomic) transaction into this
+        transaction's subledger, creating the subledger on first use.
+
+        The subledger sits between the base and the working view: from then
+        on view() reads through it, this transaction's own changes are made
+        on top of it, apply() commits it with them and discard() drops it.
+        Before each inner transaction the working view's pending changes are
+        moved into the subledger, so that the subledger is never modified
+        underneath them.
+
+        The inner transaction is applied with tapATOMIC_EMIT to a closed
+        child view of the subledger, where it runs its own preflight,
+        preclaim, hooks and invariant checks. It is merged into the subledger
+        only if it was applied (tes or tec) and this transaction can still
+        be applied on top of it: its account must exist exactly when it did
+        before, its sequence or ticket must still be valid and its balance
+        must still cover its fee.
+
+        RNG (featureRNG): there are two subledgers once any inner
+        transaction's Hooks draw from the RNG. The visible subledger is the
+        one the strong Hooks see (view() reads through it, the returned entry
+        describes it); its inner transactions draw from a shadow RNG forked
+        off the real one. The real subledger applies the same inner
+        transactions with the real RNG, and is the one committed. Nothing
+        about the real subledger reaches the caller: the result describes the
+        visible application only. An inner transaction that drew, and every
+        one applied after it, is flagged stfIndeterminate. Until the first
+        draw both subledgers are the same, and only one exists.
+    */
+    SubledgerResult
+    applyToSubledger(std::shared_ptr<STTx const> const& stx);
+
+    /** End of the strong phase (featureAtomicEmit): continue on the real
+        subledger.
+
+        Called once every strong Hook chain has run, before their state is
+        finalized: nothing that runs afterwards can roll this transaction
+        back, so from here on the working view sits on the real subledger,
+        and the visible one is dropped.
+
+        The working view's pending changes, made on top of the visible
+        subledger, are merged into the real one field by field (see
+        subledger::mergeEntry). Returns false, leaving everything in place,
+        if the subledgers cannot be reconciled: an inner transaction applied
+        to the visible subledger was not applied to the real one, this
+        transaction could not be applied after the real subledger, or a merge
+        conflicted. The caller must then fail the transaction
+        (tecSUBLEDGER_DIVERGED), which discards both.
+
+        Returns true and does nothing when there is no real subledger (no
+        inner transaction drew from the RNG, or there is no subledger).
+    */
+    bool
+    concretizeSubledger();
+
+    /** True iff this transaction has a subledger. */
+    bool
+    hasSubledger() const
+    {
+        return static_cast<bool>(subledger_);
+    }
+
+    /** The RNG cursors Hooks executing for this transaction draw from
+        instead of the Random object, or nullptr if they draw from the Random
+        object directly (see SubledgerRng). For an inner transaction
+        (tapATOMIC_EMIT) these are its parent's, and `shadow` says which
+        cursor; otherwise they exist while this transaction has a subledger
+        and `shadow` is false. */
+    SubledgerRng*
+    subledgerRng(bool& shadow) const;
+
+    /** True iff a Hook drew from this transaction's real RNG cursor. */
+    bool
+    subledgerRngDrawn() const
+    {
+        return rng_ && rng_->realDrawn;
+    }
+
+    /** The subledger's transactions, in application order. Each is a
+        SubledgerTransaction { EmittedTxnID, EmittedTxn,
+        TransactionMetaData }. */
+    std::vector<std::shared_ptr<STObject const>> const&
+    subledgerEntries() const
+    {
+        return subledgerEntries_;
+    }
+
+    /** Get the number of unapplied changes made by this transaction.
+
+        With a subledger this also counts the entries the subledger changes,
+        since apply() commits them as part of this transaction's metadata.
+        An entry changed by both is counted twice, which is conservative
+        (the count only feeds the oversizeMetaDataCap checks). */
     std::size_t
     size();
 
-    /** Visit unapplied changes. */
+    /** Visit unapplied changes. With a subledger, these are this
+        transaction's own changes and `before` is read from the subledger. */
     void
     visit(std::function<void(
               uint256 const& key,
@@ -110,7 +249,7 @@ public:
     TxMeta
     generateProvisionalMeta()
     {
-        return view_->generateProvisionalMeta(base_, tx, journal);
+        return view_->generateProvisionalMeta(viewBase(), tx, journal);
     }
 
     /** Applies all invariant checkers one by one.
@@ -145,9 +284,57 @@ private:
         XRPAmount const fee,
         std::index_sequence<Is...>);
 
+    // the view the working view sits on: the subledger if there is one,
+    // otherwise the base
+    OpenView&
+    viewBase()
+    {
+        return subledger_ ? *subledger_ : base_;
+    }
+
+    // true iff this transaction can still be applied on top of `after`
+    bool
+    stillApplies(ReadView const& after) const;
+
+    std::optional<TxMeta>
+    commitSubledger(TER ter);
+
+    // copy subledgerEntries_ into the working view's Subledger metadata
+    void
+    syncSubledgerMetaData();
+
+    // merge the working view's pending changes, made on top of the visible
+    // subledger, into the real subledger; false on a conflict
+    bool
+    mergeInto(OpenView& real);
+
+    // write the real RNG cursor to the working view's Random object
+    void
+    writeBackRng();
+
+    std::shared_ptr<STObject const>
+    makeSubledgerEntry(STTx const& stx, TxMeta const& meta, bool indeterminate)
+        const;
+
     OpenView& base_;
     ApplyFlags flags_;
     std::optional<ApplyViewImpl> view_;
+    // see applyToSubledger(). subledger_ is the subledger view() reads
+    // through: until concretizeSubledger() the visible one, afterwards the
+    // real one.
+    std::unique_ptr<OpenView> subledger_;
+    std::vector<std::shared_ptr<STObject const>> subledgerEntries_;
+    // the real subledger, once it differs from the visible one (an inner
+    // transaction drew from the RNG); its entries
+    std::unique_ptr<OpenView> realSubledger_;
+    std::vector<std::shared_ptr<STObject const>> realEntries_;
+    // an inner transaction drew from the RNG: it and everything applied
+    // after it are indeterminate
+    bool indeterminate_ = false;
+    // the real subledger could not follow the visible one
+    bool diverged_ = false;
+    // see SubledgerRng; exists while subledger_ does
+    std::unique_ptr<SubledgerRng> rng_;
 };
 
 }  // namespace ripple

src/xrpld/app/tx/detail/Transactor.cpp

@@ -94,6 +94,18 @@ preflight0(PreflightContext const& ctx)
 NotTEC
 preflight1(PreflightContext const& ctx)
 {
+    // emit_atomic inner txn (tapATOMIC_EMIT): the flag bypasses the signature
+    // checks below and in checkSign, so refuse anything carrying it that is
+    // not an emitted txn or that carries signature material (same gate as
+    // checkSign, mirroring the SF_EMITTED/tapPREFLIGHT_EMIT pair).
+    if ((ctx.flags & tapATOMIC_EMIT) &&
+        (!hook::isEmittedTxn(ctx.tx) || !hook::hasNoSignatureMaterial(ctx.tx)))
+    {
+        JLOG(ctx.j.warn())
+            << "preflight1: tapATOMIC_EMIT on a non-emitted or signed txn";
+        return temINVALID;
+    }
+
     // This is inappropriate in preflight0, because only Change transactions
     // skip this function, and those do not allow an sfTicketSequence field.
     if (ctx.tx.isFieldPresent(sfTicketSequence) &&
@@ -128,7 +140,7 @@ preflight1(PreflightContext const& ctx)
     {
         if ((ctx.app.getHashRouter().getFlags(ctx.tx.getTransactionID()) &
              SF_EMITTED) ||
-            (ctx.flags & tapPREFLIGHT_EMIT))
+            (ctx.flags & tapPREFLIGHT_EMIT) || (ctx.flags & tapATOMIC_EMIT))
         {
             if (ctx.tx.getSeqProxy().isTicket() &&
                 ctx.tx.isFieldPresent(sfAccountTxnID))
@@ -706,7 +718,10 @@ Transactor::checkPriorTxAndLastLedger(PreclaimContext const& ctx)
     if (ctx.view.txExists(ctx.tx.getTransactionID()))
         return tefALREADY;
 
-    if (hook::isEmittedTxn(ctx.tx) && ctx.view.rules().enabled(featureHooks))
+    // emit_atomic inner txns never have an emitted directory entry: they are
+    // applied by their parent, not injected from the directory.
+    if (hook::isEmittedTxn(ctx.tx) && ctx.view.rules().enabled(featureHooks) &&
+        !(ctx.flags & tapATOMIC_EMIT))
     {
         // check if the emitted txn exists on ledger and is in the emission
         // directory if not that's a re-apply so discard
@@ -885,6 +900,16 @@ Transactor::apply()
 NotTEC
 Transactor::checkSign(PreclaimContext const& ctx)
 {
+    // emit_atomic inner txn: signature verification is bypassed here, so
+    // repeat the preflight1 gate.
+    if (ctx.flags & tapATOMIC_EMIT)
+    {
+        if (!hook::isEmittedTxn(ctx.tx) ||
+            !hook::hasNoSignatureMaterial(ctx.tx))
+            return temINVALID;
+        return tesSUCCESS;
+    }
+
     // hook emitted transactions do not have signatures
     if (ctx.view.rules().enabled(featureHooks) && hook::isEmittedTxn(ctx.tx))
     {
@@ -1537,6 +1562,16 @@ Transactor::doHookCallback(
     if (!ctx_.tx.isFieldPresent(sfEmitDetails))
         return;
 
+    // emit_atomic inner txns never run their callback (the field is still
+    // present because etxn_details() adds it for any hook with a cbak): the
+    // emitting hook inspects the result itself (subledger_slot), and the
+    // subledger can still be abandoned after this point.
+    if (ctx_.flags() & tapATOMIC_EMIT)
+    {
+        JLOG(j_.trace()) << "HookInfo: callback skipped for emit_atomic txn";
+        return;
+    }
+
     auto const& emitDetails = const_cast<ripple::STTx&>(ctx_.tx)
                                   .getField(sfEmitDetails)
                                   .downcast<STObject>();
@@ -2022,7 +2057,7 @@ Transactor::operator()()
     // here despite not being emitted here by a hook here.
     if ((ctx_.flags() & tapPREFLIGHT_EMIT) ||
         (view().flags() & tapPREFLIGHT_EMIT) ||
-        (ctx_.isEmittedTxn() &&
+        (ctx_.isEmittedTxn() && !(ctx_.flags() & tapATOMIC_EMIT) &&
          !(ctx_.app.getHashRouter().getFlags(ctx_.tx.getTransactionID()) &
            SF_EMITTED)))
         return {tecINTERNAL, false};
@@ -2082,6 +2117,14 @@ Transactor::operator()()
             result = doTSH(true, tsh, stateMap, hookResults, {});
         }
 
+        // featureAtomicEmit: the strong phase is over, nothing that runs from
+        // here on can roll this transaction back, so continue on the real
+        // subledger (the visible one only existed for the strong Hooks).
+        // Everything below, including the state the strong Hooks set, is
+        // applied on top of the real one.
+        if (isTesSuccess(result) && !ctx_.concretizeSubledger())
+            result = tecSUBLEDGER_DIVERGED;
+
         // write state if all chains executed successfully
         if (isTesSuccess(result))
         {
@@ -2165,8 +2208,11 @@ Transactor::finishApply(
     // in place for the next transaction in the pass, and would draw again on
     // the final pass against a different state. Draws only exist when the
     // amendment is enabled, so this needs no separate gate.
+    // featureAtomicEmit: while there is a subledger, draws go to the real
+    // cursor (SubledgerRng) instead; an inner transaction's draws count.
     bool const rngDrawn =
-        dynamic_cast<ApplyViewImpl&>(ctx_.view()).getRandomData().has_value();
+        dynamic_cast<ApplyViewImpl&>(ctx_.view()).getRandomData().has_value() ||
+        ctx_.subledgerRngDrawn();
 
     if (isTecClaim(result) && (view().flags() & tapFAIL_HARD))
     {
@@ -2537,6 +2583,25 @@ Transactor::finishApply(
             result = tecOVERSIZE;
     }
 
+    // A subledger (emit_atomic) is only ever committed with a transaction
+    // that succeeds (ApplyContext::commitSubledger asserts this). The
+    // weak-phase oversize check above turns an applied transaction into a tec
+    // without resetting it; when it has a subledger, reset it now like any
+    // other tec, which drops the subledger and claims the fee only.
+    if (applied && !isTesSuccess(result) && ctx_.hasSubledger())
+    {
+        auto const resetResult = reset(fee);
+        if (!isTesSuccess(resetResult.first))
+            result = resetResult.first;
+        fee = resetResult.second;
+
+        if (isTesSuccess(result) || isTecClaim(result))
+            result = ctx_.checkInvariants(result, fee);
+
+        if (!isTecClaim(result) && !isTesSuccess(result))
+            applied = false;
+    }
+
     std::optional<TxMeta> metadata;
     if (applied)
     {

src/xrpld/app/tx/detail/apply.cpp

@@ -62,6 +62,12 @@ checkValidity(
         {
             // pass, this is a txn being preflighted emit api
         }
+        else if (applyFlags & tapATOMIC_EMIT)
+        {
+            // pass, this is an emit_atomic inner txn being applied into its
+            // parent's subledger (preflight2 reaches here for every non-dry-run
+            // txn)
+        }
         else if (flags & SF_EMITTED)
         {
             // pass, this txn came out of the emission directory

src/xrpld/ledger/ApplyView.h

@@ -46,7 +46,11 @@ enum ApplyFlags : std::uint32_t {
 
     // Transaction shouldn't be applied
     // Signatures shouldn't be checked
-    tapDRY_RUN = 0x1000
+    tapDRY_RUN = 0x1000,
+
+    // emit_atomic inner txn applied into its parent's subledger. Only ever
+    // set by ApplyContext::applyToSubledger.
+    tapATOMIC_EMIT = 0x2000,
 };
 
 constexpr ApplyFlags

src/xrpld/ledger/ApplyViewImpl.h

@@ -107,6 +107,15 @@ public:
         hookEmission_ = std::move(emissions);
     }
 
+    /** Set the Subledger metadata: one SubledgerTransaction for each
+        atomically emitted transaction applied as part of this one.
+    */
+    void
+    setSubledgerMetaData(std::vector<STObject>&& subledger)
+    {
+        subledger_ = std::move(subledger);
+    }
+
     void
     copyHookMetaData(
         std::vector<STObject>& execution /* in */,
@@ -164,6 +173,7 @@ private:
     std::vector<STObject> hookExecution_;
     std::vector<STObject> hookEmission_;
     std::optional<uint256> randomData_;
+    std::vector<STObject> subledger_;
 };
 
 }  // namespace ripple

src/xrpld/ledger/OpenView.h

@@ -29,6 +29,7 @@
 #include <boost/container/pmr/polymorphic_allocator.hpp>
 
 #include <functional>
+#include <optional>
 #include <utility>
 
 namespace ripple {
@@ -46,6 +47,80 @@ struct open_ledger_t
 
 extern open_ledger_t const open_ledger;
 
+/** Closed nested view construction tag.
+
+    A view constructed with this tag sits on top of another OpenView
+    and is always treated as a closed ledger, regardless of the base.
+    It is used to apply one atomically emitted hook transaction
+    (emit_atomic) on top of a subledger, so that the transaction can be
+    checked before it is merged into the subledger or thrown away.
+    See ApplyContext::applyToSubledger.
+*/
+struct closed_view_t
+{
+    explicit closed_view_t() = default;
+};
+
+extern closed_view_t const closed_view;
+
+/** Subledger construction tag.
+
+    A view constructed with this tag is the subledger of a transaction
+    whose hooks emitted transactions atomically (emit_atomic). It sits on
+    top of the view that transaction is being applied to and reports the
+    same open() as that view. Its own transactions are numbered from zero
+    and txExists() also consults the base. See ApplyContext.
+*/
+struct subledger_view_t
+{
+    explicit subledger_view_t() = default;
+};
+
+extern subledger_view_t const subledger_view;
+
+/** RNG cursors of a transaction that has a subledger (featureAtomicEmit).
+
+    While a transaction has a subledger, no Hook draws from the ledger's
+    Random object directly. Every draw goes through one of two cursors:
+
+    real    the RNG as it will be committed. Seeded from the Random object
+            on first use; it carries on from any draw made before the
+            subledger existed. The owning transaction's own Hooks, and the
+            real application of every inner transaction, draw from it. It is
+            written back to the Random object when the transaction is
+            committed or discarded, so draws are never undone.
+
+    shadow  forked off `real` (one-way) on the first draw made while an inner
+            transaction is applied to the subledger the emitting Hooks can
+            see. Outcomes drawn from it are never committed.
+
+    Neither cursor is ever stored in a ledger entry, so no inner transaction
+    modifies the Random object and none of its metadata can expose the state.
+    See ApplyContext::applyToSubledger.
+*/
+struct SubledgerRng
+{
+    explicit SubledgerRng(uint256 const& otxnID_, bool open_)
+        : otxnID(otxnID_), open(open_)
+    {
+    }
+
+    // the transaction that owns the subledger (input to the shadow fork)
+    uint256 const otxnID;
+    // whether that transaction is applied to an open ledger: inner
+    // transactions are applied to closed views, but their Hooks may draw on
+    // the same seed as the owning transaction (see HookAPI::rng_next)
+    bool const open;
+
+    std::optional<uint256> real;
+    std::optional<uint256> shadow;
+
+    // something drew from `real` (it must be written back)
+    bool realDrawn = false;
+    // something drew from `shadow` since this was last cleared
+    bool shadowDrawn = false;
+};
+
 //------------------------------------------------------------------------------
 
 /** Writable ledger view that accumulates state and tx changes.
@@ -98,6 +173,18 @@ private:
     std::shared_ptr<void const> hold_;
     bool open_ = true;
 
+    // closed_view: number of transactions already in the base chain (so
+    // TransactionIndex continues from the base). closed_view and
+    // subledger_view: the base OpenView for txExists() delegation.
+    // Zero / nullptr for every other view.
+    std::size_t baseTxCount_ = 0;
+    OpenView const* baseTxs_ = nullptr;
+
+    // closed_view only: the RNG cursors Hooks of the transaction applied to
+    // this view draw from, and which one (see SubledgerRng)
+    SubledgerRng* rng_ = nullptr;
+    bool rngShadow_ = false;
+
 public:
     OpenView() = delete;
     OpenView&
@@ -170,6 +257,60 @@ public:
     */
     OpenView(ReadView const* base, std::shared_ptr<void const> hold = nullptr);
 
+    /** Construct a sandbox view on top of another OpenView.
+
+        Effects:
+
+            The LedgerInfo and rules are copied from the base
+            (the sequence is NOT incremented).
+
+            The view always reports closed (open() == false), even
+            when the base is an open ledger, so that transactions
+            applied into it see consensus-side semantics.
+
+            txCount() continues from the base so that metadata
+            TransactionIndex values stay contiguous, and txExists()
+            also consults the base.
+
+            When `rng` is given, Hooks executed by the transaction applied to
+            this view draw from its shadow cursor if `shadow` is true and
+            from its real cursor otherwise, never from the Random object.
+    */
+    OpenView(
+        closed_view_t,
+        OpenView const& base,
+        SubledgerRng* rng = nullptr,
+        bool shadow = false);
+
+    /** The RNG cursors of a closed_view child (see SubledgerRng), or nullptr.
+     */
+    SubledgerRng*
+    subledgerRng() const
+    {
+        return rng_;
+    }
+
+    /** True iff draws in this view use the shadow cursor. */
+    bool
+    subledgerRngShadow() const
+    {
+        return rngShadow_;
+    }
+
+    /** Construct a subledger on top of another OpenView.
+
+        Effects:
+
+            The LedgerInfo, rules and open() are copied from the base
+            (the sequence is NOT incremented).
+
+            txCount() counts only this view's own transactions, so the
+            transactions applied into it are numbered from zero.
+
+            txExists() also consults the base.
+    */
+    OpenView(subledger_view_t, OpenView const& base);
+
     /** Returns true if this reflects an open ledger. */
     bool
     open() const override
@@ -189,6 +330,24 @@ public:
     void
     apply(TxsRawView& to) const;
 
+    /** Apply state changes only (no transactions).
+
+        Also forwards the XRP destroyed in this view
+        (RawStateTable::apply calls to.rawDestroyXRP).
+        Used to fold a subledger into the transaction that owns
+        it: the subledger's transactions are recorded in that
+        transaction's metadata, not in the ledger's tx list.
+    */
+    void
+    applyState(RawView& to) const;
+
+    /** The number of state entries this view changes in its base. */
+    std::size_t
+    stateSize() const
+    {
+        return items_.size();
+    }
+
     // ReadView
 
     LedgerInfo const&

src/xrpld/ledger/detail/ApplyStateTable.cpp

@@ -51,6 +51,14 @@ ApplyStateTable::apply(RawView& to) const
     }
 }
 
+void
+ApplyStateTable::clear()
+{
+    items_.clear();
+    dropsDestroyed_ = XRPAmount{0};
+    originalThreadingState_.clear();
+}
+
 std::size_t
 ApplyStateTable::size() const
 {
@@ -118,6 +126,7 @@ ApplyStateTable::generateTxMeta(
     std::optional<STAmount> const& deliver,
     std::vector<STObject> const& hookExecution,
     std::vector<STObject> const& hookEmission,
+    std::vector<STObject> const& subledger,
     beast::Journal j,
     bool isProvisional)
 {
@@ -131,6 +140,9 @@ ApplyStateTable::generateTxMeta(
     if (!hookEmission.empty())
         meta.setHookEmissions(STArray{hookEmission, sfHookEmissions});
 
+    if (!subledger.empty())
+        meta.setSubledger(STArray{subledger, sfSubledger});
+
     Mods newMod;
     for (auto& item : items_)
     {
@@ -309,6 +321,7 @@ ApplyStateTable::apply(
     std::vector<STObject> const& hookExecution,
     std::vector<STObject> const& hookEmission,
     std::optional<uint256> const& randomData,
+    std::vector<STObject> const& subledger,
     bool isDryRun,
     beast::Journal j)
 {
@@ -320,8 +333,8 @@ ApplyStateTable::apply(
     if (!to.open() || isDryRun)
     {
         // generate meta
-        auto [meta, newMod] =
-            generateTxMeta(to, tx, deliver, hookExecution, hookEmission, j);
+        auto [meta, newMod] = generateTxMeta(
+            to, tx, deliver, hookExecution, hookEmission, subledger, j);
 
         // featureRNG: added only here, never to provisional metadata, so a
         // weak Hook reading meta_slot cannot learn the RNG state.

src/xrpld/ledger/detail/ApplyStateTable.h

@@ -79,6 +79,10 @@ public:
     void
     apply(RawView& to) const;
 
+    /** Forget every buffered change, cached entry and destroyed drop. */
+    void
+    clear();
+
     std::pair<TxMeta, Mods>
     generateTxMeta(
         OpenView const& to,
@@ -86,6 +90,7 @@ public:
         std::optional<STAmount> const& deliver,
         std::vector<STObject> const& hookExecution,
         std::vector<STObject> const& hookEmission,
+        std::vector<STObject> const& subledger,
         beast::Journal j,
         bool isProvisional = false);
 
@@ -98,6 +103,7 @@ public:
         std::vector<STObject> const& hookExecution,
         std::vector<STObject> const& hookEmission,
         std::optional<uint256> const& randomData,
+        std::vector<STObject> const& subledger,
         bool isDryRun,
         beast::Journal j);
 

src/xrpld/ledger/detail/ApplyViewBase.cpp

@@ -19,6 +19,7 @@
 
 #include <xrpld/ledger/detail/ApplyViewBase.h>
 #include <xrpl/basics/contract.h>
+#include <xrpl/beast/utility/instrumentation.h>
 #include <xrpl/protocol/STAccount.h>
 
 namespace ripple {
@@ -175,6 +176,26 @@ ApplyViewBase::rawDestroyXRP(XRPAmount const& fee)
     items_.destroyXRP(fee);
 }
 
+void
+ApplyViewBase::flush(RawView& to)
+{
+    items_.apply(to);
+    items_.clear();
+}
+
+void
+ApplyViewBase::rebase(ReadView const* base)
+{
+    // size() excludes entries that were only cached by read-only peek()
+    // calls; clear() drops those too, as they belong to the old base
+    XRPL_ASSERT(
+        items_.size() == 0,
+        "ripple::detail::ApplyViewBase::rebase : nothing buffered (flush "
+        "first)");
+    items_.clear();
+    base_ = base;
+}
+
 std::map<std::tuple<AccountID, AccountID, Currency>, STAmount>
 ApplyViewBase::balanceChanges(ReadView const& view) const
 {

src/xrpld/ledger/detail/ApplyViewBase.h

@@ -126,6 +126,21 @@ public:
     std::map<std::tuple<AccountID, AccountID, Currency>, STAmount>
     balanceChanges(ReadView const& view) const;
 
+    /** Move every pending change (and the XRP destroyed so far) into `to`,
+        leaving this view with nothing buffered. Entries obtained from
+        peek() before the call must not be passed to update() or erase()
+        afterwards.
+    */
+    void
+    flush(RawView& to);
+
+    /** Continue on top of a different base. Nothing may be buffered: call
+        flush() first. Used to layer a transaction's working view on its
+        subledger and back (see ApplyContext).
+    */
+    void
+    rebase(ReadView const* base);
+
 protected:
     ApplyFlags flags_;
     ReadView const* base_;

src/xrpld/ledger/detail/ApplyViewImpl.cpp

@@ -44,6 +44,7 @@ ApplyViewImpl::apply(
         hookExecution_,
         hookEmission_,
         randomData_,
+        subledger_,
         isDryRun,
         j);
 }
@@ -60,6 +61,7 @@ ApplyViewImpl::generateProvisionalMeta(
         deliver_,
         hookExecution_,
         hookEmission_,
+        subledger_,
         j,
         true);  // isProvisional = true
     return meta;

src/xrpld/ledger/detail/OpenView.cpp

@@ -23,6 +23,8 @@
 namespace ripple {
 
 open_ledger_t const open_ledger{};
+closed_view_t const closed_view{};
+subledger_view_t const subledger_view{};
 
 class OpenView::txs_iter_impl : public txs_type::iter_base
 {
@@ -86,7 +88,11 @@ OpenView::OpenView(OpenView const& rhs)
     , base_{rhs.base_}
     , items_{rhs.items_}
     , hold_{rhs.hold_}
-    , open_{rhs.open_} {};
+    , open_{rhs.open_}
+    , baseTxCount_{rhs.baseTxCount_}
+    , baseTxs_{rhs.baseTxs_}
+    , rng_{rhs.rng_}
+    , rngShadow_{rhs.rngShadow_} {};
 
 OpenView::OpenView(
     open_ledger_t,
@@ -120,10 +126,35 @@ OpenView::OpenView(ReadView const* base, std::shared_ptr<void const> hold)
 {
 }
 
+OpenView::OpenView(
+    closed_view_t,
+    OpenView const& base,
+    SubledgerRng* rng,
+    bool shadow)
+    : OpenView(&base)
+{
+    rng_ = rng;
+    rngShadow_ = rng != nullptr && shadow;
+    // Always a closed view: inner (atomically emitted) transactions must
+    // see the same rules as during consensus ledger construction even when
+    // the parent is being applied against the open ledger.
+    open_ = false;
+    baseTxCount_ = base.txCount();
+    baseTxs_ = &base;
+}
+
+OpenView::OpenView(subledger_view_t, OpenView const& base) : OpenView(&base)
+{
+    // open() is inherited from the base: the transaction that owns the
+    // subledger is applied on top of it and keeps its own semantics. The
+    // subledger's transactions are numbered from zero.
+    baseTxs_ = &base;
+}
+
 std::size_t
 OpenView::txCount() const
 {
-    return txs_.size();
+    return baseTxCount_ + txs_.size();
 }
 
 void
@@ -134,6 +165,12 @@ OpenView::apply(TxsRawView& to) const
         to.rawTxInsert(item.first, item.second.txn, item.second.meta);
 }
 
+void
+OpenView::applyState(RawView& to) const
+{
+    items_.apply(to);
+}
+
 //---
 
 LedgerInfo const&
@@ -207,7 +244,12 @@ OpenView::txsEnd() const -> std::unique_ptr<txs_type::iter_base>
 bool
 OpenView::txExists(key_type const& key) const
 {
-    return txs_.find(key) != txs_.end();
+    if (txs_.find(key) != txs_.end())
+        return true;
+    // closed_view / subledger_view: a duplicate of a transaction already in
+    // the base view must be detected here (tefALREADY) rather than in
+    // rawTxInsert (LogicError) when the view is applied to its base.
+    return baseTxs_ != nullptr && baseTxs_->txExists(key);
 }
 
 auto

src/xrpld/ledger/detail/RawStateTable.h

@@ -89,6 +89,13 @@ public:
     void
     destroyXRP(XRPAmount const& fee);
 
+    /** The number of state entries this table would change in its base. */
+    std::size_t
+    size() const
+    {
+        return items_.size();
+    }
+
     std::unique_ptr<ReadView::sles_type::iter_base>
     slesBegin(ReadView const& base) const;