diff --git a/src/ledger/LedgerManager.h b/src/ledger/LedgerManager.h
index e10f11ab3d..fdef9705d7 100644
--- a/src/ledger/LedgerManager.h
+++ b/src/ledger/LedgerManager.h
@@ -7,6 +7,7 @@
 #include "catchup/LedgerApplyManager.h"
 #include "history/HistoryManager.h"
 #include "ledger/NetworkConfig.h"
+#include "rust/RustBridge.h"
 #include <memory>
 
 namespace stellar
@@ -200,6 +201,12 @@ class LedgerManager
     virtual void manuallyAdvanceLedgerHeader(LedgerHeader const& header) = 0;
 
     virtual SorobanMetrics& getSorobanMetrics() = 0;
+    virtual rust_bridge::SorobanModuleCache& getModuleCache() = 0;
+
+    // Compiles all contracts in the current ledger, for ledger protocols
+    // starting at minLedgerVersion and running through to
+    // Config::CURRENT_LEDGER_PROTOCOL_VERSION (to enable upgrades).
+    virtual void compileAllContractsInLedger(uint32_t minLedgerVersion) = 0;
 
     virtual ~LedgerManager()
     {
diff --git a/src/ledger/LedgerManagerImpl.cpp b/src/ledger/LedgerManagerImpl.cpp
index 2bea36c459..74c524977c 100644
--- a/src/ledger/LedgerManagerImpl.cpp
+++ b/src/ledger/LedgerManagerImpl.cpp
@@ -22,6 +22,7 @@
 #include "ledger/LedgerTxn.h"
 #include "ledger/LedgerTxnEntry.h"
 #include "ledger/LedgerTxnHeader.h"
+#include "ledger/SharedModuleCacheCompiler.h"
 #include "main/Application.h"
 #include "main/Config.h"
 #include "main/ErrorMessages.h"
@@ -42,6 +43,7 @@
 #include "work/WorkScheduler.h"
 #include "xdrpp/printer.h"
 
+#include <cstdint>
 #include <fmt/format.h>
 
 #include "xdr/Stellar-ledger.h"
@@ -155,6 +157,7 @@ LedgerManagerImpl::LedgerManagerImpl(Application& app)
     , mCatchupDuration(
           app.getMetrics().NewTimer({"ledger", "catchup", "duration"}))
     , mState(LM_BOOTING_STATE)
+    , mModuleCache(::rust_bridge::new_module_cache())
 
 {
     setupLedgerCloseMetaStream();
@@ -403,6 +406,9 @@ LedgerManagerImpl::loadLastKnownLedger(bool restoreBucketlist)
         updateNetworkConfig(ltx);
         mSorobanNetworkConfigReadOnly = mSorobanNetworkConfigForApply;
     }
+
+    // Prime module cache with ledger content.
+    compileAllContractsInLedger(latestLedgerHeader->ledgerVersion);
 }
 
 Database&
@@ -566,6 +572,27 @@ LedgerManagerImpl::getSorobanMetrics()
     return mSorobanMetrics;
 }
 
+rust_bridge::SorobanModuleCache&
+LedgerManagerImpl::getModuleCache()
+{
+    return *mModuleCache;
+}
+
+void
+LedgerManagerImpl::compileAllContractsInLedger(uint32_t minLedgerVersion)
+{
+    auto& moduleCache = getModuleCache();
+    std::vector<uint32_t> ledgerVersions;
+    for (uint32_t i = minLedgerVersion;
+         i <= Config::CURRENT_LEDGER_PROTOCOL_VERSION; i++)
+    {
+        ledgerVersions.push_back(i);
+    }
+    auto compiler = std::make_shared<SharedModuleCacheCompiler>(
+        mApp, moduleCache, ledgerVersions);
+    compiler->run();
+}
+
 void
 LedgerManagerImpl::publishSorobanMetrics()
 {
diff --git a/src/ledger/LedgerManagerImpl.h b/src/ledger/LedgerManagerImpl.h
index 3538021a04..d11a23685b 100644
--- a/src/ledger/LedgerManagerImpl.h
+++ b/src/ledger/LedgerManagerImpl.h
@@ -11,6 +11,7 @@
 #include "ledger/NetworkConfig.h"
 #include "ledger/SorobanMetrics.h"
 #include "main/PersistentState.h"
+#include "rust/RustBridge.h"
 #include "transactions/TransactionFrame.h"
 #include "util/XDRStream.h"
 #include "xdr/Stellar-ledger.h"
@@ -152,6 +153,9 @@ class LedgerManagerImpl : public LedgerManager
     // Update cached ledger state values managed by this class.
     void advanceLedgerPointers(CloseLedgerOutput const& output);
 
+    // The reusable / inter-ledger soroban module cache.
+    ::rust::Box<rust_bridge::SorobanModuleCache> mModuleCache;
+
   protected:
     // initialLedgerVers must be the ledger version at the start of the ledger
     // and currLedgerVers is the ledger version in the current ltx header. These
@@ -251,5 +255,7 @@ class LedgerManagerImpl : public LedgerManager
     {
         return mCurrentlyApplyingLedger;
     }
+    rust_bridge::SorobanModuleCache& getModuleCache() override;
+    void compileAllContractsInLedger(uint32_t minLedgerVersion) override;
 };
 }
diff --git a/src/ledger/SharedModuleCacheCompiler.cpp b/src/ledger/SharedModuleCacheCompiler.cpp
new file mode 100644
index 0000000000..7a17f1caa2
--- /dev/null
+++ b/src/ledger/SharedModuleCacheCompiler.cpp
@@ -0,0 +1,161 @@
+#include "ledger/SharedModuleCacheCompiler.h"
+#include "bucket/SearchableBucketList.h"
+#include "crypto/Hex.h"
+#include "crypto/SHA.h"
+#include "main/Application.h"
+#include "rust/RustBridge.h"
+#include "util/Logging.h"
+#include <chrono>
+
+using namespace stellar;
+
+SharedModuleCacheCompiler::SharedModuleCacheCompiler(
+    Application& app, rust_bridge::SorobanModuleCache& moduleCache,
+    std::vector<uint32_t> const& ledgerVersions)
+    : mApp(app)
+    , mModuleCache(moduleCache)
+    , mSnap(app.getBucketManager()
+                .getBucketSnapshotManager()
+                .copySearchableLiveBucketListSnapshot())
+    , mNumThreads(
+          static_cast<size_t>(std::max(2, app.getConfig().WORKER_THREADS) - 1))
+    , mLedgerVersions(ledgerVersions)
+{
+}
+
+void
+SharedModuleCacheCompiler::pushWasm(xdr::xvector<uint8_t> const& vec)
+{
+    std::unique_lock<std::mutex> lock(mMutex);
+    mHaveSpace.wait(
+        lock, [&] { return mBytesLoaded - mBytesCompiled < MAX_MEM_BYTES; });
+    xdr::xvector<uint8_t> buf(vec);
+    auto size = buf.size();
+    mWasms.emplace_back(std::move(buf));
+    mBytesLoaded += size;
+    lock.unlock();
+    mHaveContracts.notify_all();
+    LOG_INFO(DEFAULT_LOG, "Loaded contract with {} bytes of wasm code", size);
+}
+
+bool
+SharedModuleCacheCompiler::isFinishedCompiling(
+    std::unique_lock<std::mutex>& lock)
+{
+    releaseAssert(lock.owns_lock());
+    return mLoadedAll && mBytesCompiled == mBytesLoaded;
+}
+
+void
+SharedModuleCacheCompiler::setFinishedLoading()
+{
+    std::unique_lock lock(mMutex);
+    mLoadedAll = true;
+    lock.unlock();
+    mHaveContracts.notify_all();
+}
+
+bool
+SharedModuleCacheCompiler::popAndCompileWasm(size_t thread,
+                                             std::unique_lock<std::mutex>& lock)
+{
+
+    releaseAssert(lock.owns_lock());
+
+    // Wait for a new contract to compile (or being done).
+    mHaveContracts.wait(
+        lock, [&] { return !mWasms.empty() || isFinishedCompiling(lock); });
+
+    // Check to see if we were woken up due to end-of-compilation.
+    if (isFinishedCompiling(lock))
+    {
+        return false;
+    }
+
+    xdr::xvector<uint8_t> wasm = std::move(mWasms.front());
+    mWasms.pop_front();
+
+    // Make a local shallow copy of the cache, so we don't race on the
+    // shared host.
+    auto cache = mModuleCache.shallow_clone();
+
+    lock.unlock();
+
+    auto start = std::chrono::steady_clock::now();
+    auto slice = rust::Slice<const uint8_t>(wasm.data(), wasm.size());
+    try
+    {
+        for (auto ledgerVersion : mLedgerVersions)
+        {
+            cache->compile(ledgerVersion, slice);
+        }
+    }
+    catch (std::exception const& e)
+    {
+        LOG_ERROR(DEFAULT_LOG, "Thread {} failed to compile wasm code: {}",
+                  thread, e.what());
+    }
+    auto end = std::chrono::steady_clock::now();
+    auto dur_us =
+        std::chrono::duration_cast<std::chrono::microseconds>(end - start);
+    LOG_INFO(DEFAULT_LOG, "Thread {} compiled {} byte wasm contract {} in {}us",
+             thread, wasm.size(), binToHex(sha256(wasm)), dur_us.count());
+    lock.lock();
+    mTotalCompileTime += dur_us;
+    mBytesCompiled += wasm.size();
+    wasm.clear();
+    mHaveSpace.notify_all();
+    mHaveContracts.notify_all();
+    return true;
+}
+
+void
+SharedModuleCacheCompiler::run()
+{
+    auto self = shared_from_this();
+    auto start = std::chrono::steady_clock::now();
+    LOG_INFO(DEFAULT_LOG,
+             "Launching 1 loading and {} compiling background threads",
+             mNumThreads);
+    mApp.postOnBackgroundThread(
+        [self]() {
+            self->mSnap->scanForContractCode([&](LedgerEntry const& entry) {
+                self->pushWasm(entry.data.contractCode().code);
+                return Loop::INCOMPLETE;
+            });
+            self->setFinishedLoading();
+        },
+        "contract loading thread");
+
+    for (auto thread = 0; thread < self->mNumThreads; ++thread)
+    {
+        mApp.postOnBackgroundThread(
+            [self, thread]() {
+                size_t nContractsCompiled = 0;
+                std::unique_lock<std::mutex> lock(self->mMutex);
+                while (!self->isFinishedCompiling(lock))
+                {
+                    if (self->popAndCompileWasm(thread, lock))
+                    {
+                        ++nContractsCompiled;
+                    }
+                }
+                LOG_INFO(DEFAULT_LOG, "Thread {} compiled {} contracts", thread,
+                         nContractsCompiled);
+            },
+            fmt::format("compilation thread {}", thread));
+    }
+
+    std::unique_lock lock(self->mMutex);
+    self->mHaveContracts.wait(
+        lock, [self, &lock] { return self->isFinishedCompiling(lock); });
+
+    auto end = std::chrono::steady_clock::now();
+    LOG_INFO(DEFAULT_LOG,
+             "All contracts compiled in {}ms real time, {}ms CPU time",
+             std::chrono::duration_cast<std::chrono::milliseconds>(end - start)
+                 .count(),
+             std::chrono::duration_cast<std::chrono::milliseconds>(
+                 self->mTotalCompileTime)
+                 .count());
+}
diff --git a/src/ledger/SharedModuleCacheCompiler.h b/src/ledger/SharedModuleCacheCompiler.h
new file mode 100644
index 0000000000..4151f20cb7
--- /dev/null
+++ b/src/ledger/SharedModuleCacheCompiler.h
@@ -0,0 +1,59 @@
+#pragma once
+// Copyright 2024 Stellar Development Foundation and contributors. Licensed
+// under the Apache License, Version 2.0. See the COPYING file at the root
+// of this distribution or at http://www.apache.org/licenses/LICENSE-2.0
+
+#include "bucket/BucketSnapshotManager.h"
+#include "rust/RustBridge.h"
+#include "xdrpp/types.h"
+
+#include <cstdint>
+#include <deque>
+
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+
+namespace stellar
+{
+class Application;
+}
+
+// This class encapsulates a multithreaded strategy for loading contracts
+// out of the database (on one thread) and compiling them (on N-1 others).
+class SharedModuleCacheCompiler
+    : public std::enable_shared_from_this<SharedModuleCacheCompiler>
+{
+    stellar::Application& mApp;
+    stellar::rust_bridge::SorobanModuleCache& mModuleCache;
+    stellar::SearchableSnapshotConstPtr mSnap;
+    std::deque<xdr::xvector<uint8_t>> mWasms;
+
+    const size_t mNumThreads;
+    const size_t MAX_MEM_BYTES = 10 * 1024 * 1024;
+    bool mLoadedAll{false};
+    size_t mBytesLoaded{0};
+    size_t mBytesCompiled{0};
+    std::vector<uint32_t> mLedgerVersions;
+
+    std::mutex mMutex;
+    std::condition_variable mHaveSpace;
+    std::condition_variable mHaveContracts;
+
+    std::chrono::microseconds mTotalCompileTime{0};
+
+    void setFinishedLoading();
+    bool isFinishedCompiling(std::unique_lock<std::mutex>& lock);
+    // This gets called in a loop on the loader/producer thread.
+    void pushWasm(xdr::xvector<uint8_t> const& vec);
+    // This gets called in a loop on the compiler/consumer threads. It returns
+    // true if anything was actually compiled.
+    bool popAndCompileWasm(size_t thread, std::unique_lock<std::mutex>& lock);
+
+  public:
+    SharedModuleCacheCompiler(
+        stellar::Application& app,
+        stellar::rust_bridge::SorobanModuleCache& moduleCache,
+        std::vector<uint32_t> const& ledgerVersions);
+    void run();
+};
diff --git a/src/main/ApplicationUtils.cpp b/src/main/ApplicationUtils.cpp
index 67158b0cc0..54295c25b3 100644
--- a/src/main/ApplicationUtils.cpp
+++ b/src/main/ApplicationUtils.cpp
@@ -23,6 +23,7 @@
 #include "main/PersistentState.h"
 #include "main/StellarCoreVersion.h"
 #include "overlay/OverlayManager.h"
+#include "rust/RustBridge.h"
 #include "scp/LocalNode.h"
 #include "util/GlobalChecks.h"
 #include "util/Logging.h"
@@ -1034,4 +1035,18 @@ listContracts(Config const& cfg)
     return 0;
 }
 
+int
+compileContracts(Config const& cfg)
+{
+    VirtualClock clock(VirtualClock::REAL_TIME);
+    auto config = cfg;
+    config.setNoListen();
+    auto app = Application::create(clock, config, /* newDB */ false);
+    // Initializing the ledgerManager will, in restoring the last known ledger,
+    // also cause all contracts to be compiled. All we need to do is start and
+    // stop the application.
+    app->start();
+    return 0;
+}
+
 }
diff --git a/src/main/ApplicationUtils.h b/src/main/ApplicationUtils.h
index c9af6ae634..8bb11c00d0 100644
--- a/src/main/ApplicationUtils.h
+++ b/src/main/ApplicationUtils.h
@@ -66,4 +66,5 @@ std::optional<uint32_t>
 getStellarCoreMajorReleaseVersion(std::string const& vstr);
 
 int listContracts(Config const& cfg);
+int compileContracts(Config const& cfg);
 }
diff --git a/src/main/CommandLine.cpp b/src/main/CommandLine.cpp
index b637d5c0df..771842193a 100644
--- a/src/main/CommandLine.cpp
+++ b/src/main/CommandLine.cpp
@@ -1478,8 +1478,15 @@ runListContracts(CommandLineArgs const& args)
 
     return runWithHelp(args, {configurationParser(configOption)},
                        [&] { return listContracts(configOption.getConfig()); });
+}
 
+int
+runCompileContracts(CommandLineArgs const& args)
+{
+    CommandLine::ConfigOption configOption;
 
+    return runWithHelp(args, {configurationParser(configOption)}, [&] {
+        return compileContracts(configOption.getConfig());
     });
 }
 
@@ -1960,6 +1967,8 @@ handleCommandLine(int argc, char* const* argv)
          {"list-contracts",
           "List sha256 hashes of all contract code entries in the bucket list",
           runListContracts},
+         {"compile-contracts", "Compile wasm files and cache them",
+          runCompileContracts},
          {"version", "print version information", runVersion}}};
 
     auto adjustedCommandLine = commandLine.adjustCommandLine({argc, argv});
diff --git a/src/rust/Cargo.toml b/src/rust/Cargo.toml
index 08af378b52..3dc581bfff 100644
--- a/src/rust/Cargo.toml
+++ b/src/rust/Cargo.toml
@@ -79,16 +79,16 @@ tracy-client = { version = "=0.17.0", features = [
 # number grow gradually is also not the end of the world.
 
 # [dependencies.soroban-env-host-p22]
-# version = "=22.0.0"
+# version = "=22.1.3"
 # git = "https://github.com/stellar/rs-soroban-env"
 # package = "soroban-env-host"
-# rev = "0497816694bef2b103494c8c61b7c8a06a72c7d3"
+# rev = "ad0d6d27dab5a1711eaaf8af8783fdf203742eb3"
 
 # [dependencies.soroban-env-host-p21]
-# version = "=21.2.0"
+# version = "=21.2.2"
 # git = "https://github.com/stellar/rs-soroban-env"
 # package = "soroban-env-host"
-# rev = "8809852dcf8489f99407a5ceac12625ee3d14693"
+# rev = "7eeddd897cfb0f700f938b0c8d6f0541150d1fcb"
 
 # The test wasms and synth-wasm crate should usually be taken from the highest
 # supported host, since test material usually just grows over time.
diff --git a/src/rust/soroban/p23 b/src/rust/soroban/p23
index 822727b37b..ad0d6d27da 160000
--- a/src/rust/soroban/p23
+++ b/src/rust/soroban/p23
@@ -1 +1 @@
-Subproject commit 822727b37b7ef2eea1fc0bafc558820dc450c67e
+Subproject commit ad0d6d27dab5a1711eaaf8af8783fdf203742eb3
diff --git a/src/rust/src/contract.rs b/src/rust/src/contract.rs
index 358ba69f53..a7710a7ed9 100644
--- a/src/rust/src/contract.rs
+++ b/src/rust/src/contract.rs
@@ -10,7 +10,7 @@ use crate::{
         InvokeHostFunctionOutput, RustBuf, SorobanVersionInfo, XDRFileHash,
     },
 };
-use log::{debug, trace, warn};
+use log::{debug, error, trace, warn};
 use std::{fmt::Display, io::Cursor, panic, rc::Rc, time::Instant};
 
 // This module (contract) is bound to _two separate locations_ in the module
@@ -18,8 +18,8 @@ use std::{fmt::Display, io::Cursor, panic, rc::Rc, time::Instant};
 // hi) version-specific definition of stellar_env_host. We therefore
 // import it from our _parent_ module rather than from the crate root.
 pub(crate) use super::soroban_env_host::{
-    budget::Budget,
-    e2e_invoke::{self, extract_rent_changes, LedgerEntryChange},
+    budget::{AsBudget, Budget},
+    e2e_invoke::{extract_rent_changes, LedgerEntryChange},
     fees::{
         compute_rent_fee as host_compute_rent_fee,
         compute_transaction_resource_fee as host_compute_transaction_resource_fee,
@@ -32,8 +32,9 @@ pub(crate) use super::soroban_env_host::{
         LedgerEntryExt, Limits, ReadXdr, ScError, ScErrorCode, ScErrorType, ScSymbol, ScVal,
         TransactionEnvelope, TtlEntry, WriteXdr, XDR_FILES_SHA256,
     },
-    HostError, LedgerInfo, VERSION,
+    HostError, LedgerInfo, Val, VERSION,
 };
+use super::{ErrorHandler, ModuleCache};
 use std::error::Error;
 
 impl TryFrom<&CxxLedgerInfo> for LedgerInfo {
@@ -336,6 +337,7 @@ pub(crate) fn invoke_host_function(
     ttl_entries: &Vec<CxxBuf>,
     base_prng_seed: &CxxBuf,
     rent_fee_configuration: &CxxRentFeeConfiguration,
+    module_cache: &crate::SorobanModuleCache,
 ) -> Result<InvokeHostFunctionOutput, Box<dyn Error>> {
     let res = panic::catch_unwind(panic::AssertUnwindSafe(|| {
         invoke_host_function_or_maybe_panic(
@@ -350,6 +352,7 @@ pub(crate) fn invoke_host_function(
             ttl_entries,
             base_prng_seed,
             rent_fee_configuration,
+            module_cache,
         )
     }));
     match res {
@@ -405,6 +408,7 @@ fn invoke_host_function_or_maybe_panic(
     ttl_entries: &Vec<CxxBuf>,
     base_prng_seed: &CxxBuf,
     rent_fee_configuration: &CxxRentFeeConfiguration,
+    module_cache: &crate::SorobanModuleCache,
 ) -> Result<InvokeHostFunctionOutput, Box<dyn Error>> {
     #[cfg(feature = "tracy")]
     let client = tracy_client::Client::start();
@@ -432,7 +436,8 @@ fn invoke_host_function_or_maybe_panic(
     let (res, time_nsecs) = {
         let _span1 = tracy_span!("e2e_invoke::invoke_function");
         let start_time = Instant::now();
-        let res = e2e_invoke::invoke_host_function_with_trace_hook(
+
+        let res = super::invoke_host_function_with_trace_hook_and_module_cache(
             &budget,
             enable_diagnostics,
             hf_buf,
@@ -445,6 +450,7 @@ fn invoke_host_function_or_maybe_panic(
             base_prng_seed,
             &mut diagnostic_events,
             trace_hook,
+            module_cache,
         );
         let stop_time = Instant::now();
         let time_nsecs = stop_time.duration_since(start_time).as_nanos() as u64;
@@ -630,3 +636,92 @@ pub(crate) fn can_parse_transaction(xdr: &CxxBuf, depth_limit: u32) -> bool {
     ));
     res.is_ok()
 }
+
+#[allow(dead_code)]
+#[derive(Clone)]
+struct CoreCompilationContext {
+    unlimited_budget: Budget,
+}
+
+impl super::CompilationContext for CoreCompilationContext {}
+
+#[allow(dead_code)]
+impl CoreCompilationContext {
+    fn new() -> Result<Self, Box<dyn std::error::Error>> {
+        let unlimited_budget = Budget::try_from_configs(
+            u64::MAX,
+            u64::MAX,
+            ContractCostParams(vec![].try_into().unwrap()),
+            ContractCostParams(vec![].try_into().unwrap()),
+        )?;
+        Ok(CoreCompilationContext { unlimited_budget })
+    }
+}
+
+impl AsBudget for CoreCompilationContext {
+    fn as_budget(&self) -> &Budget {
+        &self.unlimited_budget
+    }
+}
+
+impl ErrorHandler for CoreCompilationContext {
+    fn map_err<T, E>(&self, res: Result<T, E>) -> Result<T, HostError>
+    where
+        super::soroban_env_host::Error: From<E>,
+        E: core::fmt::Debug,
+    {
+        match res {
+            Ok(t) => Ok(t),
+            Err(e) => {
+                error!("compiling module: {:?}", e);
+                Err(HostError::from(e))
+            }
+        }
+    }
+
+    fn error(&self, error: super::soroban_env_host::Error, msg: &str, _args: &[Val]) -> HostError {
+        error!("compiling module: {:?}: {}", error, msg);
+        HostError::from(error)
+    }
+}
+
+#[allow(dead_code)]
+pub(crate) struct ProtocolSpecificModuleCache {
+    compilation_context: CoreCompilationContext,
+    pub(crate) module_cache: ModuleCache,
+}
+
+#[allow(dead_code)]
+impl ProtocolSpecificModuleCache {
+    pub(crate) fn new() -> Result<Self, Box<dyn std::error::Error>> {
+        let compilation_context = CoreCompilationContext::new()?;
+        let module_cache = ModuleCache::new_reusable(&compilation_context)?;
+        Ok(ProtocolSpecificModuleCache {
+            compilation_context,
+            module_cache,
+        })
+    }
+
+    pub(crate) fn compile(&mut self, wasm: &[u8]) -> Result<(), Box<dyn std::error::Error>> {
+        Ok(self.module_cache.parse_and_cache_module_simple(
+            &self.compilation_context,
+            get_max_proto(),
+            wasm,
+        )?)
+    }
+
+    // This produces a new `SorobanModuleCache` with a separate
+    // `CoreCompilationContext` but a clone of the underlying `ModuleCache`, which
+    // will (since the module cache is the reusable flavor) actually point to
+    // the _same_ underlying threadsafe map of `Module`s and the same associated
+    // `Engine` as those that `self` currently points to.
+    //
+    // This mainly exists to allow cloning a shared-ownership handle to a
+    // (threadsafe) ModuleCache to pass to separate C++-launched threads, to
+    // allow multithreaded compilation.
+    pub(crate) fn shallow_clone(&self) -> Result<Self, Box<dyn std::error::Error>> {
+        let mut new = Self::new()?;
+        new.module_cache = self.module_cache.clone();
+        Ok(new)
+    }
+}
diff --git a/src/rust/src/lib.rs b/src/rust/src/lib.rs
index 1df96ef9a0..a98cfa4cbc 100644
--- a/src/rust/src/lib.rs
+++ b/src/rust/src/lib.rs
@@ -192,6 +192,7 @@ mod rust_bridge {
             ttl_entries: &Vec<CxxBuf>,
             base_prng_seed: &CxxBuf,
             rent_fee_configuration: CxxRentFeeConfiguration,
+            module_cache: &SorobanModuleCache,
         ) -> Result<InvokeHostFunctionOutput>;
 
         fn init_logging(maxLevel: LogLevel) -> Result<()>;
@@ -263,6 +264,16 @@ mod rust_bridge {
             xdr: &CxxBuf,
             depth_limit: u32,
         ) -> Result<bool>;
+
+        type SorobanModuleCache;
+
+        fn new_module_cache() -> Result<Box<SorobanModuleCache>>;
+        fn compile(
+            self: &mut SorobanModuleCache,
+            ledger_protocol: u32,
+            source: &[u8],
+        ) -> Result<()>;
+        fn shallow_clone(self: &SorobanModuleCache) -> Result<Box<SorobanModuleCache>>;
     }
 
     // And the extern "C++" block declares C++ stuff we're going to import to
@@ -522,27 +533,118 @@ use log::partition::TX;
 mod p23 {
     pub(crate) extern crate soroban_env_host_p23;
     pub(crate) use soroban_env_host_p23 as soroban_env_host;
+    use super::SorobanModuleCache;
+    use soroban_env_host::{
+        budget::Budget,
+        e2e_invoke::{self, InvokeHostFunctionResult},
+        xdr::DiagnosticEvent,
+        HostError, LedgerInfo, TraceHook,
+    };
 
     pub(crate) mod contract;
 
+    // We do some more local re-exports here of things used in contract.rs that
+    // don't exist in older hosts (eg. the p21 & 22 hosts, where we define stubs for
+    // these imports).
+    pub(crate) use soroban_env_host::{CompilationContext, ErrorHandler, ModuleCache};
+
     // An adapter for some API breakage between p21 and p22.
     pub(crate) const fn get_version_pre_release(v: &soroban_env_host::Version) -> u32 {
         v.interface.pre_release
     }
 
-    pub(crate) const fn get_version_protocol(v: &soroban_env_host::Version) -> u32 {
+    pub(crate) const fn get_version_protocol(_v: &soroban_env_host::Version) -> u32 {
         // Temporarily hardcode the protocol version until we actually bump it
         // in the host library.
         23
     }
+
+    pub fn invoke_host_function_with_trace_hook_and_module_cache<
+    T: AsRef<[u8]>,
+    I: ExactSizeIterator<Item = T>,
+>(
+    budget: &Budget,
+    enable_diagnostics: bool,
+    encoded_host_fn: T,
+    encoded_resources: T,
+    encoded_source_account: T,
+    encoded_auth_entries: I,
+    ledger_info: LedgerInfo,
+    encoded_ledger_entries: I,
+    encoded_ttl_entries: I,
+    base_prng_seed: T,
+    diagnostic_events: &mut Vec<DiagnosticEvent>,
+    trace_hook: Option<TraceHook>,
+    module_cache: &SorobanModuleCache,
+) -> Result<InvokeHostFunctionResult, HostError> {
+    e2e_invoke::invoke_host_function_with_trace_hook_and_module_cache(
+        &budget,
+        enable_diagnostics,
+        encoded_host_fn,
+        encoded_resources,
+        encoded_source_account,
+        encoded_auth_entries,
+        ledger_info,
+        encoded_ledger_entries,
+        encoded_ttl_entries,
+        base_prng_seed,
+        diagnostic_events,
+        trace_hook,
+        Some(module_cache.p23_cache.module_cache.clone()),
+    )
+}
+
 }
 
 #[path = "."]
 mod p22 {
     pub(crate) extern crate soroban_env_host_p22;
     pub(crate) use soroban_env_host_p22 as soroban_env_host;
-
     pub(crate) mod contract;
+    use super::SorobanModuleCache;
+    use soroban_env_host::{
+        budget::{AsBudget,Budget},
+        e2e_invoke::{self, InvokeHostFunctionResult},
+        xdr::DiagnosticEvent,
+        Error, HostError, LedgerInfo, TraceHook, Val
+    };
+
+    // Some stub definitions to handle API additions for the
+    // reusable module cache.
+
+    #[allow(dead_code)]
+    const INTERNAL_ERROR: Error = Error::from_type_and_code(
+        soroban_env_host::xdr::ScErrorType::Context,
+        soroban_env_host::xdr::ScErrorCode::InternalError,
+    );
+
+    #[allow(dead_code)]
+    #[derive(Clone)]
+    pub(crate) struct ModuleCache;
+    #[allow(dead_code)]
+    pub(crate) trait ErrorHandler {
+        fn map_err<T, E>(&self, res: Result<T, E>) -> Result<T, HostError>
+        where
+            Error: From<E>,
+            E: core::fmt::Debug;
+        fn error(&self, error: Error, msg: &str, args: &[Val]) -> HostError;
+    }
+    #[allow(dead_code)]
+    impl ModuleCache {
+        pub(crate) fn new_reusable<T>(_handler: T) -> Result<Self, HostError> {
+            Err(INTERNAL_ERROR.into())
+        }
+        pub(crate) fn parse_and_cache_module_simple<T>(
+            &self,
+            _handler: &T,
+            _protocol: u32,
+            _wasm: &[u8],
+        ) -> Result<(), HostError> {
+            Err(INTERNAL_ERROR.into())
+        }
+    }
+    #[allow(dead_code)]
+    pub(crate) trait CompilationContext: ErrorHandler + AsBudget {}
 
     // An adapter for some API breakage between p21 and p22.
     pub(crate) const fn get_version_pre_release(v: &soroban_env_host::Version) -> u32 {
@@ -552,14 +654,91 @@ mod p22 {
     pub(crate) const fn get_version_protocol(v: &soroban_env_host::Version) -> u32 {
         v.interface.protocol
     }
+
+    pub fn invoke_host_function_with_trace_hook_and_module_cache<
+        T: AsRef<[u8]>,
+        I: ExactSizeIterator<Item = T>,
+    >(
+        budget: &Budget,
+        enable_diagnostics: bool,
+        encoded_host_fn: T,
+        encoded_resources: T,
+        encoded_source_account: T,
+        encoded_auth_entries: I,
+        ledger_info: LedgerInfo,
+        encoded_ledger_entries: I,
+        encoded_ttl_entries: I,
+        base_prng_seed: T,
+        diagnostic_events: &mut Vec<DiagnosticEvent>,
+        trace_hook: Option<TraceHook>,
+        _module_cache: &SorobanModuleCache,
+    ) -> Result<InvokeHostFunctionResult, HostError> {
+        e2e_invoke::invoke_host_function_with_trace_hook(
+            &budget,
+            enable_diagnostics,
+            encoded_host_fn,
+            encoded_resources,
+            encoded_source_account,
+            encoded_auth_entries,
+            ledger_info,
+            encoded_ledger_entries,
+            encoded_ttl_entries,
+            base_prng_seed,
+            diagnostic_events,
+            trace_hook,
+        )
+    }
 }
 
 #[path = "."]
 mod p21 {
     pub(crate) extern crate soroban_env_host_p21;
     pub(crate) use soroban_env_host_p21 as soroban_env_host;
-
     pub(crate) mod contract;
+    use super::SorobanModuleCache;
+    use soroban_env_host::{
+        budget::{AsBudget, Budget},
+        e2e_invoke::{self, InvokeHostFunctionResult},
+        xdr::DiagnosticEvent,
+        Error, HostError, LedgerInfo, TraceHook, Val,
+    };
+
+    // Some stub definitions to handle API additions for the
+    // reusable module cache.
+
+    #[allow(dead_code)]
+    const INTERNAL_ERROR: Error = Error::from_type_and_code(
+        soroban_env_host::xdr::ScErrorType::Context,
+        soroban_env_host::xdr::ScErrorCode::InternalError,
+    );
+
+    #[allow(dead_code)]
+    #[derive(Clone)]
+    pub(crate) struct ModuleCache;
+    #[allow(dead_code)]
+    pub(crate) trait ErrorHandler {
+        fn map_err<T, E>(&self, res: Result<T, E>) -> Result<T, HostError>
+        where
+            Error: From<E>,
+            E: core::fmt::Debug;
+        fn error(&self, error: Error, msg: &str, args: &[Val]) -> HostError;
+    }
+    #[allow(dead_code)]
+    impl ModuleCache {
+        pub(crate) fn new_reusable<T>(_handler: T) -> Result<Self, HostError> {
+            Err(INTERNAL_ERROR.into())
+        }
+        pub(crate) fn parse_and_cache_module_simple<T>(
+            &self,
+            _handler: &T,
+            _protocol: u32,
+            _wasm: &[u8],
+        ) -> Result<(), HostError> {
+            Err(INTERNAL_ERROR.into())
+        }
+    }
+    #[allow(dead_code)]
+    pub(crate) trait CompilationContext: ErrorHandler + AsBudget {}
 
     // An adapter for some API breakage between p21 and p22.
     pub(crate) const fn get_version_pre_release(v: &soroban_env_host::Version) -> u32 {
@@ -569,6 +748,40 @@ mod p21 {
     pub(crate) const fn get_version_protocol(v: &soroban_env_host::Version) -> u32 {
         soroban_env_host::meta::get_ledger_protocol_version(v.interface)
     }
+
+    pub fn invoke_host_function_with_trace_hook_and_module_cache<
+        T: AsRef<[u8]>,
+        I: ExactSizeIterator<Item = T>,
+    >(
+        budget: &Budget,
+        enable_diagnostics: bool,
+        encoded_host_fn: T,
+        encoded_resources: T,
+        encoded_source_account: T,
+        encoded_auth_entries: I,
+        ledger_info: LedgerInfo,
+        encoded_ledger_entries: I,
+        encoded_ttl_entries: I,
+        base_prng_seed: T,
+        diagnostic_events: &mut Vec<DiagnosticEvent>,
+        trace_hook: Option<TraceHook>,
+        _module_cache: &SorobanModuleCache,
+    ) -> Result<InvokeHostFunctionResult, HostError> {
+        e2e_invoke::invoke_host_function_with_trace_hook(
+            &budget,
+            enable_diagnostics,
+            encoded_host_fn,
+            encoded_resources,
+            encoded_source_account,
+            encoded_auth_entries,
+            ledger_info,
+            encoded_ledger_entries,
+            encoded_ttl_entries,
+            base_prng_seed,
+            diagnostic_events,
+            trace_hook,
+        )
+    }
 }
 
 // We alias the latest soroban as soroban_curr to help reduce churn in code
@@ -650,6 +863,7 @@ struct HostModule {
         ttl_entries: &Vec<CxxBuf>,
         base_prng_seed: &CxxBuf,
         rent_fee_configuration: &CxxRentFeeConfiguration,
+        module_cache: &SorobanModuleCache,
     ) -> Result<InvokeHostFunctionOutput, Box<dyn std::error::Error>>,
     compute_transaction_resource_fee:
         fn(tx_resources: CxxTransactionResources, fee_config: CxxFeeConfiguration) -> FeePair,
@@ -747,6 +961,7 @@ pub(crate) fn invoke_host_function(
     ttl_entries: &Vec<CxxBuf>,
     base_prng_seed: &CxxBuf,
     rent_fee_configuration: CxxRentFeeConfiguration,
+    module_cache: &SorobanModuleCache,
 ) -> Result<InvokeHostFunctionOutput, Box<dyn std::error::Error>> {
     let hm = get_host_module_for_protocol(config_max_protocol, ledger_info.protocol_version)?;
     let res = (hm.invoke_host_function)(
@@ -761,6 +976,7 @@ pub(crate) fn invoke_host_function(
         ttl_entries,
         base_prng_seed,
         &rent_fee_configuration,
+        module_cache,
     );
 
     #[cfg(feature = "testutils")]
@@ -779,6 +995,7 @@ pub(crate) fn invoke_host_function(
         ttl_entries,
         base_prng_seed,
         rent_fee_configuration,
+        module_cache,
     );
 
     res
@@ -810,6 +1027,7 @@ mod test_extra_protocol {
         ttl_entries: &Vec<CxxBuf>,
         base_prng_seed: &CxxBuf,
         rent_fee_configuration: CxxRentFeeConfiguration,
+        module_cache: &SorobanModuleCache,
     ) {
         if let Ok(extra) = std::env::var("SOROBAN_TEST_EXTRA_PROTOCOL") {
             if let Ok(proto) = u32::from_str(&extra) {
@@ -841,6 +1059,7 @@ mod test_extra_protocol {
                         ttl_entries,
                         base_prng_seed,
                         &rent_fee_configuration,
+                        module_cache,
                     );
                     if mostly_the_same_host_function_output(&res1, &res2) {
                         info!(target: TX, "{}", summarize_host_function_output(hm1, &res1));
@@ -1025,3 +1244,52 @@ pub(crate) fn compute_write_fee_per_1kb(
     let hm = get_host_module_for_protocol(config_max_protocol, protocol_version)?;
     Ok((hm.compute_write_fee_per_1kb)(bucket_list_size, fee_config))
 }
+
+// The SorobanModuleCache needs to hold a different protocol-specific cache for
+// each supported protocol version it's going to be used with. It has to hold
+// all these caches _simultaneously_ because it might perform an upgrade from
+// protocol N to protocol N+1 in a single transaction, and needs to be ready for
+// that before it happens.
+//
+// Most of these caches can be empty at any given time, because we're not
+// expecting core to need to replay old protocols, and/or if it does it's during
+// replay and there's no problem stalling while filling a cache with new entries
+// on a per-ledger basis as they are replayed.
+//
+// But for the current protocol version we need to have a cache ready to execute
+// anything thrown at it once it's in sync, so we should prime the
+// current-protocol cache as soon as we start, as well as the next-protocol
+// cache (if it exists) so that we can upgrade without stalling.
+struct SorobanModuleCache {
+    p23_cache: p23::contract::ProtocolSpecificModuleCache,
+}
+
+impl SorobanModuleCache {
+    fn new() -> Result<Self, Box<dyn std::error::Error>> {
+        Ok(Self {
+            p23_cache: p23::contract::ProtocolSpecificModuleCache::new()?,
+        })
+    }
+    pub fn compile(
+        &mut self,
+        ledger_protocol: u32,
+        wasm: &[u8],
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        match ledger_protocol {
+            23 => self.p23_cache.compile(wasm),
+            // Add other protocols here as needed.
+            _ => Err(Box::new(soroban_curr::contract::CoreHostError::General(
+                "unsupported protocol",
+            ))),
+        }
+    }
+    pub fn shallow_clone(&self) -> Result<Box<Self>, Box<dyn std::error::Error>> {
+        Ok(Box::new(Self {
+            p23_cache: self.p23_cache.shallow_clone()?,
+        }))
+    }
+}
+
+fn new_module_cache() -> Result<Box<SorobanModuleCache>, Box<dyn std::error::Error>> {
+    Ok(Box::new(SorobanModuleCache::new()?))
+}
diff --git a/src/transactions/InvokeHostFunctionOpFrame.cpp b/src/transactions/InvokeHostFunctionOpFrame.cpp
index ae76c16264..51adfccb3a 100644
--- a/src/transactions/InvokeHostFunctionOpFrame.cpp
+++ b/src/transactions/InvokeHostFunctionOpFrame.cpp
@@ -493,7 +493,8 @@ InvokeHostFunctionOpFrame::doApply(
             toCxxBuf(getSourceID()), authEntryCxxBufs,
             getLedgerInfo(ltx, app, sorobanConfig), ledgerEntryCxxBufs,
             ttlEntryCxxBufs, basePrngSeedBuf,
-            sorobanConfig.rustBridgeRentFeeConfiguration());
+            sorobanConfig.rustBridgeRentFeeConfiguration(),
+            app.getLedgerManager().getModuleCache());
         metrics.mCpuInsn = out.cpu_insns;
         metrics.mMemByte = out.mem_bytes;
         metrics.mInvokeTimeNsecs = out.time_nsecs;