emit atomic subledger concept

-

hook/error.h

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

hook/extern.h

@@ -339,6 +339,23 @@ prepare(
     uint32_t read_ptr,
     uint32_t read_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/sfcodes.h

@@ -299,6 +299,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)
@@ -327,3 +328,4 @@
 #define sfActiveValidators ((15U << 16U) + 95U)
 #define sfGenesisMints ((15U << 16U) + 96U)
 #define sfRemarks ((15U << 16U) + 97U)
+#define sfSubledger ((15U << 16U) + 98U)

include/xrpl/hook/Enum.h

@@ -17,6 +17,7 @@
 #define featureHooksUpdate2 "1"
 #define fix20250131 "1"
 #define fixGuardDepth32 "1"
+#define featureAtomicEmit "1"
 namespace hook_api {
 struct Rules
 {
@@ -385,7 +386,9 @@ enum class hook_return_code : int64_t {
     MEM_OVERLAP = -43,   // one or more specified buffers are the same memory
     TOO_MANY_STATE_MODIFICATIONS = -44,  // more than 256 modified state
                                          // entires in the combined hook chains
-    TOO_MANY_NAMESPACES = -45
+    TOO_MANY_NAMESPACES = -45,
+    ALREADY_IN_SUBLEDGER = -46  // emit_atomic called by a hook executing
+                                // inside a subledger (an emit_atomic txn)
 };
 
 enum class ExitType : uint8_t {
@@ -399,6 +402,10 @@ const uint16_t max_state_modifications = 256;
 const uint8_t max_slots = 255;
 const uint8_t max_nonce = 255;
 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

@@ -372,3 +372,18 @@ HOOK_API_DEFINITION(
 HOOK_API_DEFINITION(
   int64_t, prepare, (uint32_t, uint32_t, uint32_t, uint32_t),
   featureHooksUpdate2)
+
+// int64_t emit_atomic(uint32_t write_ptr, uint32_t write_len, uint32_t read_ptr, uint32_t read_len);
+HOOK_API_DEFINITION(
+  int64_t, emit_atomic, (uint32_t, uint32_t, uint32_t, uint32_t),
+  featureAtomicEmit)
+
+// int64_t prepare_atomic(uint32_t write_ptr, uint32_t write_len, uint32_t read_ptr, uint32_t read_len);
+HOOK_API_DEFINITION(
+  int64_t, prepare_atomic, (uint32_t, uint32_t, uint32_t, uint32_t),
+  featureAtomicEmit)
+
+// int64_t subledger_slot(uint32_t slot_into, uint32_t read_ptr, uint32_t read_len);
+HOOK_API_DEFINITION(
+  int64_t, subledger_slot, (uint32_t, uint32_t, uint32_t),
+  featureAtomicEmit)

include/xrpl/protocol/TxMeta.h

@@ -152,6 +152,29 @@ public:
         return static_cast<bool>(mHookEmissions);
     }
 
+    /** 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
+    {
+        return *mSubledger;
+    }
+
+    void
+    setSubledger(STArray const& subledger)
+    {
+        mSubledger = subledger;
+    }
+
+    bool
+    hasSubledger() const
+    {
+        return static_cast<bool>(mSubledger);
+    }
+
     STAmount
     getDeliveredAmount() const
     {
@@ -176,6 +199,7 @@ private:
     std::optional<STAmount> mDelivered;
     std::optional<STArray> mHookExecutions;
     std::optional<STArray> mHookEmissions;
+    std::optional<STArray> mSubledger;
 
     STArray mNodes;
 };

include/xrpl/protocol/detail/features.macro

@@ -34,6 +34,7 @@
 // If you add an amendment here, then do not forget to increment `numFeatures`
 // in include/xrpl/protocol/Feature.h.
 
+XRPL_FEATURE(AtomicEmit,                 Supported::yes, VoteBehavior::DefaultNo)
 XRPL_FIX    (20261001,                   Supported::yes, VoteBehavior::DefaultYes)
 XRPL_FEATURE(HookOnV2_1,                 Supported::yes, VoteBehavior::DefaultNo)
 XRPL_FEATURE(OnChainManifests,           Supported::no, VoteBehavior::DefaultNo)

include/xrpl/protocol/detail/sfields.macro

@@ -393,6 +393,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
@@ -429,3 +430,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,14 @@ 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},
+         {sfEmittedTxn, soeREQUIRED},
+         {sfTransactionMetaData, soeREQUIRED}});
+
     add(sfHook.jsonName,
         sfHook.getCode(),
         {{sfHookHash, soeOPTIONAL},

src/libxrpl/protocol/TxMeta.cpp

@@ -49,6 +49,9 @@ TxMeta::TxMeta(
 
     if (obj.isFieldPresent(sfHookEmissions))
         setHookEmissions(obj.getFieldArray(sfHookEmissions));
+
+    if (obj.isFieldPresent(sfSubledger))
+        setSubledger(obj.getFieldArray(sfSubledger));
 }
 
 TxMeta::TxMeta(uint256 const& txid, std::uint32_t ledger, STObject const& obj)
@@ -75,6 +78,9 @@ TxMeta::TxMeta(uint256 const& txid, std::uint32_t ledger, STObject const& obj)
 
     if (obj.isFieldPresent(sfHookEmissions))
         setHookEmissions(obj.getFieldArray(sfHookEmissions));
+
+    if (obj.isFieldPresent(sfSubledger))
+        setSubledger(obj.getFieldArray(sfSubledger));
 }
 
 TxMeta::TxMeta(uint256 const& txid, std::uint32_t ledger, Blob const& vec)
@@ -245,6 +251,9 @@ TxMeta::getAsObject() const
     if (hasHookEmissions())
         metaData.setFieldArray(sfHookEmissions, getHookEmissions());
 
+    if (hasSubledger())
+        metaData.setFieldArray(sfSubledger, getSubledger());
+
     return metaData;
 }
 

src/test/app/HookAPI_test.cpp

@@ -19,6 +19,8 @@
 #include <test/app/Import_json.h>
 #include <test/jtx.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>
@@ -36,7 +38,11 @@ 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 +50,7 @@ private:
             tx,
             tesSUCCESS,
             env.current()->fees().base,
-            tapNONE,
+            flags,
             env.journal};
         return applyCtx;
     }
@@ -154,6 +160,536 @@ 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());
+
+            auto const entry = api.apply_atomic(*tpTrans);
+            BEAST_EXPECT(entry.has_value());
+            if (!entry)
+                return;
+            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));
+
+            // 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_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
+            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 && enabled)
+                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
@@ -4762,6 +5298,8 @@ public:
 
         test_prepare(features);
         test_emit(features);
+        test_emit_atomic(features);
+        test_emit_atomic_subledger(features);
         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

@@ -112,6 +112,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,
@@ -126,6 +127,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,123 @@ 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
+        sandbox.rawInsert(sle(7, 7));
+        sandbox.rawDestroyXRP(XRPAmount{5});
+        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 +1215,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,32 @@ 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. An atomic emission is recorded before it is executed, 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. 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;
@@ -551,6 +572,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

@@ -87,6 +87,13 @@ canHook(
     ripple::uint256 hookOn,
     std::optional<ripple::Slice> 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);
 
@@ -150,6 +157,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<
@@ -209,6 +219,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

@@ -409,7 +409,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");
@@ -449,11 +449,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))
@@ -525,7 +535,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");
@@ -534,9 +544,47 @@ 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)
+        {
+            JLOG(j.trace()) << "HookEmit[" << HC_ACC()
+                            << "]: too many emit_atomic txns for this "
+                               "transaction (max "
+                            << hook_api::max_atomic_emit << ").";
+            return Unexpected(EMISSION_FAILURE);
+        }
+    }
+
     std::shared_ptr<STTx const> stpTrans;
     try
     {
@@ -725,6 +773,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()
@@ -759,30 +827,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
@@ -839,6 +927,43 @@ 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");
+
+    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
 {
@@ -2439,6 +2564,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>
@@ -664,6 +665,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)
 {
@@ -1547,6 +1564,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
@@ -2657,12 +2679,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,
@@ -2671,6 +2692,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;
 
@@ -2680,11 +2752,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();
@@ -2698,6 +2769,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. The attempt (which may run the
+        // inner txn's hooks) is spent whatever its outcome.
+        api.recordEmission(tpTrans, true);
+        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(
@@ -2714,10 +2795,85 @@ 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. */
+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();
 }
@@ -3939,6 +4095,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>
@@ -28,6 +29,91 @@
 
 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
+
 ApplyContext::ApplyContext(
     Application& app_,
     OpenView& base,
@@ -50,12 +136,143 @@ ApplyContext::ApplyContext(
 void
 ApplyContext::discard()
 {
+    // the working view may sit on the subledger: replace it first
     view_.emplace(&base_, flags_);
+    subledger_.reset();
+    subledgerEntries_.clear();
 }
 
 std::optional<TxMeta>
 ApplyContext::apply(TER ter)
 {
+    if (subledger_)
+        return commitSubledger(ter);
+    return view_->apply(base_, tx, ter, flags_ & tapDRY_RUN, journal);
+}
+
+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();
+}
+
+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_);
+
+    // Move the working view's pending changes (e.g. the TouchCount of the
+    // TSH accounts touched so far) into the subledger and continue on top of
+    // it. The working view buffers whole entries, so nothing may change
+    // underneath it once it holds any.
+    view_->flush(*subledger_);
+    view_->rebase(subledger_.get());
+
+    auto const id = stx->getTransactionID();
+
+    // 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.
+    OpenView child(closed_view, *subledger_);
+    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
+    }
+
+    child.apply(*subledger_);
+
+    auto entry = std::make_shared<STObject>(sfSubledgerTransaction);
+    entry->setFieldH256(sfEmittedTxnID, id);
+    {
+        Serializer s;
+        stx->add(s);
+        SerialIter sit(s.slice());
+        entry->emplace_back(STObject(sit, sfEmittedTxn));
+    }
+    entry->emplace_back(result.metadata->getAsObject());
+    subledgerEntries_.push_back(entry);
+
+    JLOG(journal.trace()) << "Subledger[" << tx.getTransactionID()
+                          << "]: " << id
+                          << " applied: " << transToken(result.ter);
+    return {result.ter, std::move(entry)};
+}
+
+std::optional<TxMeta>
+ApplyContext::commitSubledger(TER ter)
+{
+    // Every path that turns a transaction with a subledger into a tec goes
+    // through discard() (see Transactor::finishApply), which drops it.
+    XRPL_ASSERT(
+        isTesSuccess(ter),
+        "ripple::ApplyContext::commitSubledger : transaction succeeded");
+
+    // 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 and delivered amount, and its metadata now
+    // describes the net change against the base.
+    view_->flush(*subledger_);
+    view_->rebase(&base_);
+    {
+        Unthreader to(*view_, base_, !base_.open());
+        subledger_->applyState(to);
+    }
+
+    std::vector<STObject> subledger;
+    subledger.reserve(subledgerEntries_.size());
+    for (auto const& entry : subledgerEntries_)
+        subledger.push_back(*entry);
+    view_->setSubledgerMetaData(std::move(subledger));
+
+    // the working view no longer refers to it
+    subledger_.reset();
+    subledgerEntries_.clear();
+
     return view_->apply(base_, tx, ter, flags_ & tapDRY_RUN, journal);
 }
 
@@ -72,7 +289,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,8 +26,10 @@
 #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 {
 
@@ -50,6 +52,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 +90,76 @@ public:
         view_->deliver(amount);
     }
 
-    /** Discard changes and start fresh. */
+    /** Discard changes and start fresh. The subledger, if any, is
+        discarded too. */
     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.
+    */
+    SubledgerResult
+    applyToSubledger(std::shared_ptr<STTx const> const& stx);
+
+    /** True iff this transaction has a subledger. */
+    bool
+    hasSubledger() const
+    {
+        return static_cast<bool>(subledger_);
+    }
+
+    /** 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
+        (changes made by its subledger's transactions are not counted). */
     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 +176,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 +211,27 @@ 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);
+
     OpenView& base_;
     ApplyFlags flags_;
     std::optional<ApplyViewImpl> view_;
+    // see applyToSubledger()
+    std::unique_ptr<OpenView> subledger_;
+    std::vector<std::shared_ptr<STObject const>> subledgerEntries_;
 };
 
 }  // 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))
@@ -704,7 +716,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
@@ -883,6 +898,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))
     {
@@ -1493,6 +1518,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>();
@@ -1955,7 +1990,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};
@@ -2050,6 +2085,16 @@ Transactor::operator()()
     if (auto stream = j_.trace())
         stream << "preclaim result: " << transToken(result);
 
+    return finishApply(result, hooksEnabled, aawMap, tsh);
+}
+
+ApplyResult
+Transactor::finishApply(
+    TER result,
+    bool hooksEnabled,
+    std::map<AccountID, std::set<uint256>>& aawMap,
+    std::vector<std::pair<AccountID, bool>>& tsh)
+{
     bool applied = isTesSuccess(result);
 
     auto fee = ctx_.tx.getFieldAmount(sfFee).xrp();
@@ -2405,6 +2450,25 @@ Transactor::operator()()
             result = tecOVERSIZE;
     }
 
+    // A subledger (emit_atomic) is only ever committed with a transaction
+    // that succeeds. 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/Transactor.h

@@ -228,6 +228,15 @@ protected:
                              // deduced until after apply i.e. pathing
                              // participants, crossed offers
 
+    // The post-apply pipeline of operator(): tec handling (reset),
+    // invariants, balance rewards, weak hooks and commit.
+    ApplyResult
+    finishApply(
+        TER result,
+        bool hooksEnabled,
+        std::map<AccountID, std::set<uint256>>& aawMap,
+        std::vector<std::pair<AccountID, bool>>& tsh);
+
     ///////////////////////////////////////////////////
 
     TER

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

@@ -57,6 +57,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 */,
@@ -148,6 +157,7 @@ private:
     std::optional<STAmount> deliver_;
     std::vector<STObject> hookExecution_;
     std::vector<STObject> hookEmission_;
+    std::vector<STObject> subledger_;
 };
 
 }  // namespace ripple

src/xrpld/ledger/OpenView.h

@@ -46,6 +46,37 @@ 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;
+
 //------------------------------------------------------------------------------
 
 /** Writable ledger view that accumulates state and tx changes.
@@ -98,6 +129,13 @@ 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;
+
 public:
     OpenView() = delete;
     OpenView&
@@ -170,6 +208,37 @@ 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.
+    */
+    OpenView(closed_view_t, OpenView const& base);
+
+    /** 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 +258,17 @@ 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;
+
     // 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_)
     {
@@ -302,6 +314,7 @@ ApplyStateTable::apply(
     std::optional<STAmount> const& deliver,
     std::vector<STObject> const& hookExecution,
     std::vector<STObject> const& hookEmission,
+    std::vector<STObject> const& subledger,
     bool isDryRun,
     beast::Journal j)
 {
@@ -313,8 +326,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);
 
         if (!isDryRun)
         {

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);
 
@@ -97,6 +102,7 @@ public:
         std::optional<STAmount> const& deliver,
         std::vector<STObject> const& hookExecution,
         std::vector<STObject> const& hookEmission,
+        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,24 @@ ApplyViewBase::rawDestroyXRP(XRPAmount const& fee)
     items_.destroyXRP(fee);
 }
 
+void
+ApplyViewBase::flush(RawView& to)
+{
+    items_.apply(to);
+    items_.clear();
+}
+
+void
+ApplyViewBase::rebase(ReadView const* base)
+{
+    XRPL_ASSERT(
+        items_.size() == 0,
+        "ripple::detail::ApplyViewBase::rebase : nothing buffered");
+    // drops entries that were only cached by read-only peek() calls
+    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

@@ -37,7 +37,15 @@ ApplyViewImpl::apply(
     beast::Journal j)
 {
     return items_.apply(
-        to, tx, ter, deliver_, hookExecution_, hookEmission_, isDryRun, j);
+        to,
+        tx,
+        ter,
+        deliver_,
+        hookExecution_,
+        hookEmission_,
+        subledger_,
+        isDryRun,
+        j);
 }
 
 TxMeta
@@ -52,6 +60,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,9 @@ 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_} {};
 
 OpenView::OpenView(
     open_ledger_t,
@@ -120,10 +124,28 @@ OpenView::OpenView(ReadView const* base, std::shared_ptr<void const> hold)
 {
 }
 
+OpenView::OpenView(closed_view_t, OpenView const& base) : OpenView(&base)
+{
+    // 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 +156,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 +235,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