[trading-native][storage-core] Commit applier + JMT pipeline + AptosDB wiring

Adds the durable-commit applier, the two-stage JMT commit pipeline,
and the AptosDB integration for native positions. Builds on the
durable storage tier in `[storage-core-db]` and the shared primitives
in `[shared] LeafEntry trait + StateSummary Option` / `[shared] Lift
async-commit pipeline`.

The in-memory scanner mirror (the `NativeStateStore` DashMap and
its readers) is layered on top in a later `[in-memory-store]`
commit; the durable + JMT side here stands on its own.

Commit applier:
- `NativeStateCommitter::apply`: per-shard `SchemaBatch` fan-out
  against `position_db` for every Position write, returning
  `NativeMerkleLeafUpdates` (`MerkleLeafUpdate` per write) for
  the JMT pipeline. Stale-index entries for prior versions are
  written into the same batch so the pruner can drop them later.

Commit pipeline:
- `PositionBufferedState` — alias of the shared
  `BufferedState<L, S, P, X>` generic with position's pipeline
  parameters and a no-op `PositionExtras`. Constructed via
  `spawn_commit_pipeline`.
- `PositionSnapshotCommitter` driver (`merklize_position`): pulls
  `PositionSnapshotToCommit` payloads, runs the snapshot through
  `ShardedJmtMerkleDb::merklize_snapshot`, forwards the resulting
  `PositionMerkleCommit` downstream.
- `PositionMerkleBatchCommitter`: pulls `PositionMerkleCommit`s and
  persists per-shard JMT batches + top-levels batch to
  `position_merkle_db`.

State-shape primitives in `storage-interface`:
`storage/storage-interface/src/state_store/sharded_jmt_state.rs`
with the JMT-sharded pipeline primitives that aren't main-state
shared: `LeafSlot<V>` + `impl LeafEntry for LeafSlot<V>`,
`ShardedJmtState<Slot>` (16 `MapLayer` chains + `extend` /
`make_delta` / `is_descendant_of`), the inherent impl on
`StateAndSummary<ShardedJmtState<Slot: LeafEntry>>`
(`new_empty`, `extend`, `make_delta`, etc.), and
`pre_shard_jmt_updates`. Adds two `StateSummary` constructors
that only JMT-sharded pipelines without a hot companion need:
`new_global_only(version, smt)` and `new_empty_global_only()`.

In-aptosdb crate:
- `PositionSlot` — type alias of `LeafSlot<()>` (value not carried
  in-slot; lives in `position_db` durably).
- `PositionStateStore` — owner / coordinator, type alias of the
  shared `PipelineStateStore<L, BS>` (also added in this commit,
  in `crate::common`). Wraps the shared `current_state` mutex
  (readable by outside consumers) and the `PositionBufferedState`
  mutex (commit-path serialization).
- `PositionDb` gains the `commit` / `commit_single_shard` /
  `write_progress` / `find_prior_version` methods that the
  committer uses.

AptosDB wiring:
- `AptosDB` gains optional handles for `PositionDb` +
  `PositionMerkleDb` + `PositionStateStore`. Accessed by the rest
  of the codebase via `native_position_handles` /
  `native_state_committer`.
- `init_native_position`: opens 16 sharded position DBs +
  unsharded merkle DB with production CF tuning. Spawns the
  async commit pipeline seeded with the empty-tree placeholder
  root. Idempotent (rejects second call). Logs the open.
- `commit_native_position` (in `aptosdb_writer.rs`): rayon-spawned
  alongside main-state commit; drains
  `output.write_set().native_position_iter()` (the WriteSet
  sibling bucket from [types]) through `NativeStateCommitter`,
  feeding the resulting `MerkleLeafUpdate`s into the position
  JMT pipeline.

Cargo.toml deps + lib.rs mod exports for the new files.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: grao1991

Cargo.lock

@@ -20,6 +20,7 @@ aptos-crypto = { workspace = true }
 aptos-db-indexer = { workspace = true }
 aptos-db-indexer-schemas = { workspace = true, features = ["fuzzing"] }
 aptos-executor-types = { workspace = true }
+aptos-experimental-layered-map = { workspace = true }
 aptos-experimental-runtimes = { workspace = true }
 aptos-infallible = { workspace = true }
 aptos-jellyfish-merkle = { workspace = true }
@@ -54,6 +55,7 @@ rayon = { workspace = true }
 serde = { workspace = true }
 static_assertions = { workspace = true }
 status-line = { workspace = true }
+thiserror = { workspace = true }
 tokio = { workspace = true }
 
 [dev-dependencies]

storage/aptosdb/Cargo.toml

@@ -449,3 +449,27 @@ where
         }
     }
 }
+
+/// `current_state` is shared with outside readers; `buffered_state` is
+/// the commit-path mutex.
+pub struct PipelineStateStore<L, BS> {
+    current_state: Arc<Mutex<L>>,
+    buffered_state: Mutex<BS>,
+}
+
+impl<L, BS> PipelineStateStore<L, BS> {
+    pub fn from_parts(current_state: Arc<Mutex<L>>, buffered_state: BS) -> Self {
+        Self {
+            current_state,
+            buffered_state: Mutex::new(buffered_state),
+        }
+    }
+
+    pub fn current_state(&self) -> Arc<Mutex<L>> {
+        Arc::clone(&self.current_state)
+    }
+
+    pub(crate) fn buffered_state_locked(&self) -> aptos_infallible::MutexGuard<'_, BS> {
+        self.buffered_state.lock()
+    }
+}

storage/aptosdb/src/common.rs

@@ -84,6 +84,9 @@ impl AptosDB {
             hot_state_config,
         ));
 
+        // The native-position DB isn't open here; it attaches later
+        // via `AptosDB::init_native_position`, which re-binds the
+        // pruner via `LedgerPrunerManager::attach_native_pruners`.
         let ledger_pruner = LedgerPrunerManager::new(
             Arc::clone(&ledger_db),
             pruner_config.ledger_pruner_config,
@@ -106,6 +109,7 @@ impl AptosDB {
             pre_commit_lock: std::sync::Mutex::new(()),
             commit_lock: std::sync::Mutex::new(()),
             update_subscriber: None,
+            position: None,
         }
     }
 
@@ -155,7 +159,7 @@ impl AptosDB {
             db.write_pruner_progress(synced_version)?;
         }
 
-        let myself = Self::new_with_dbs(
+        let mut myself = Self::new_with_dbs(
             ledger_db,
             hot_state_merkle_db,
             state_merkle_db,
@@ -169,6 +173,14 @@ impl AptosDB {
             hot_state_config,
         );
 
+        if super::ENABLE_NATIVE_POSITION {
+            myself.init_native_position(
+                db_paths,
+                rocksdb_configs.state_kv_db_config,
+                readonly,
+            )?;
+        }
+
         if !readonly {
             if let Some(version) = myself.get_synced_version()? {
                 myself

storage/aptosdb/src/db/aptosdb_internal.rs

@@ -13,6 +13,8 @@ use crate::{
     metrics::{
         COMMITTED_TXNS, LATEST_TXN_VERSION, LEDGER_VERSION, NEXT_BLOCK_EPOCH, OTHER_TIMERS_SECONDS,
     },
+    native_state_committer::NativeStateCommitter,
+    position_buffered_state::PositionProofReader,
     pruner::PrunerManager,
     schema::{
         db_metadata::{DbMetadataKey, DbMetadataSchema, DbMetadataValue},
@@ -310,11 +312,134 @@ impl AptosDB {
                     .commit_transaction_accumulator(chunk.first_version, chunk.transaction_infos)
                     .unwrap()
             });
+            // Native-position subsystem: dispatch Position writes
+            // from each TransactionOutput's `native_writes()`
+            // side-channel to `position_db` + the position JMT. Gated
+            // on the bundle being attached so baselines that never
+            // called `init_native_position` skip the spawn entirely —
+            // no rayon scheduling cost.
+            if self.position.is_some() {
+                s.spawn(|_| self.commit_native_position(chunk).unwrap());
+            }
         });
 
         Ok(new_root_hash)
     }
 
+    /// Dispatch every Position write in the block's
+    /// `TransactionOutput`s to [`NativeStateCommitter`]. No-op if the
+    /// native-position subsystem has not been initialized on this
+    /// AptosDB instance (i.e. `init_native_position` wasn't called at
+    /// node open).
+    ///
+    /// For each `output.write_set().native_position_iter()`:
+    /// 1. `NativeStateCommitter::apply` writes the Position rows to
+    ///    `position_db` (per-shard fan-out) and returns a list of
+    ///    [`crate::native_state_committer::MerkleLeafUpdate`]s for the
+    ///    position JMT.
+    /// 2. If the position async pipeline is attached, the per-version
+    ///    `ShardedJmtState::extend` updates the scratchpad SMT against
+    ///    a `PositionProofReader` and the chunk's final state is
+    ///    handed to `PipelineStateStore::buffered_state_locked()`
+    ///    + `update(..)` for the snapshot/batch committer threads —
+    ///    same two-stage shape as main state's `BufferedState`.
+    fn commit_native_position(&self, chunk: &ChunkToCommit) -> Result<()> {
+        let _timer = OTHER_TIMERS_SECONDS.timer_with(&["commit_native_position"]);
+        let Some(bundle) = self.position.as_ref() else {
+            return Ok(());
+        };
+        let committer = NativeStateCommitter::new(bundle.kv_db.clone());
+
+        // The chunk's `state.last_checkpoint().version()` is the chain-
+        // level checkpoint marker — same signal main state's
+        // `StateStore::update` uses. If it falls inside this chunk,
+        // the position chain's `last_checkpoint` must advance to the
+        // position state at that exact version. If it doesn't, we
+        // carry the prior `last_checkpoint` forward.
+        let chunk_first = chunk.first_version;
+        let chunk_last = chunk_first + chunk.len() as Version;
+        let chunk_checkpoint_version = chunk.state.last_checkpoint().version();
+        let checkpoint_within_chunk =
+            chunk_checkpoint_version.filter(|v| (chunk_first..chunk_last).contains(v));
+
+        // Snapshot the prior position state once; per-version `extend`
+        // builds the chain locally. We hand the final state to
+        // `update()` at chunk end (per-chunk granularity, matching
+        // main state).
+        let store_opt = bundle.state_store.as_ref();
+        let mut chain_state = store_opt.map(|s| s.current_state().lock().latest().clone());
+        let mut chain_checkpoint =
+            store_opt.map(|s| s.current_state().lock().last_checkpoint().clone());
+
+        let mut version = chunk_first;
+        for output in chunk.transaction_outputs {
+            // Native-position writes ride in the WriteSet's
+            // `native_positions` sibling bucket — value/hotness
+            // iterators on the same WriteSet naturally skip them, so
+            // no main-state path ever sees them.
+            let position_writes: Vec<_> = output
+                .write_set()
+                .native_position_iter()
+                .map(|(k, op)| (k.clone(), op.as_write_op().clone()))
+                .collect();
+            let merkle_updates_position = if !position_writes.is_empty() {
+                committer
+                    .apply(version, position_writes)
+                    .map_err(|e| AptosDbError::Other(format!("native commit: {e}")))?
+                    .position
+            } else {
+                Vec::new()
+            };
+
+            // Advance the per-version chain. `extend` runs the
+            // scratchpad SMT update; the `last_checkpoint` is
+            // captured below if this is the chunk's checkpoint
+            // version.
+            if let (Some(store), Some(prior_latest)) = (store_opt, chain_state.as_ref()) {
+                let proof_reader = PositionProofReader {
+                    merkle_db: bundle.merkle_db.clone(),
+                    version: store
+                        .current_state()
+                        .lock()
+                        .latest()
+                        .version()
+                        .unwrap_or(version),
+                };
+                let position_updates: Vec<_> = merkle_updates_position
+                    .into_iter()
+                    .map(|u| {
+                        (u.state_key_hash, crate::position_buffered_state::PositionSlot {
+                            state_key: u.state_key,
+                            value_hash: u.value_hash,
+                            value: None,
+                        })
+                    })
+                    .collect();
+                let new_latest = prior_latest.extend(version, position_updates, &proof_reader)?;
+                if Some(version) == checkpoint_within_chunk {
+                    chain_checkpoint = Some(new_latest.clone());
+                }
+                chain_state = Some(new_latest);
+            }
+            version += 1;
+        }
+
+        // Hand the final per-chunk state to the buffered state.
+        // `update()` enqueues a snapshot flush only if
+        // `last_checkpoint` actually advanced — matching main state.
+        if let Some(store) = store_opt {
+            let new_state =
+                crate::position_buffered_state::PositionLedgerStateWithSummary::from_latest_and_last_checkpoint(
+                    chain_state.expect("chain_state set when store is set"),
+                    chain_checkpoint.expect("chain_checkpoint set when store is set"),
+                );
+            let estimated_items = chunk.transaction_outputs.len();
+            let mut bufstate = store.buffered_state_locked();
+            bufstate.update(new_state, (), estimated_items, /* sync_commit */ false)?;
+        }
+        Ok(())
+    }
+
     fn commit_state_kv_and_ledger_metadata(&self, chunk: &ChunkToCommit) -> Result<()> {
         let _timer = OTHER_TIMERS_SECONDS.timer_with(&["commit_state_kv_and_ledger_metadata"]);
 

storage/aptosdb/src/db/aptosdb_writer.rs

@@ -2,10 +2,24 @@
 // Licensed pursuant to the Innovation-Enabling Source Code License, available at https://github.com/aptos-labs/aptos-core/blob/main/LICENSE
 
 use crate::{
-    backup::backup_handler::BackupHandler, event_store::EventStore, ledger_db::LedgerDb,
-    pruner::LedgerPrunerManager, rocksdb_property_reporter::RocksdbPropertyReporter,
-    state_kv_db::StateKvDb, state_merkle_db::StateMerkleDb, state_store::StateStore,
+    backup::backup_handler::BackupHandler,
+    event_store::EventStore,
+    ledger_db::LedgerDb,
+    native_state_committer::NativeStateCommitter,
+    position_buffered_state::new_empty_position_state,
+    position_db::{PositionDb, NUM_NATIVE_VALUE_SHARDS},
+    position_merkle_db::PositionMerkleDb,
+    position_state_store::PositionStateStore,
+    pruner::LedgerPrunerManager,
+    rocksdb_property_reporter::RocksdbPropertyReporter,
+    state_kv_db::StateKvDb,
+    state_merkle_db::StateMerkleDb,
+    state_store::StateStore,
     transaction_store::TransactionStore,
+    utils::truncation_helper::{
+        get_position_commit_progress, get_position_merkle_commit_progress,
+        truncate_position_db_shards, truncate_position_merkle_db_shards,
+    },
 };
 use aptos_config::config::{HotStateConfig, PrunerConfig, RocksdbConfigs, StorageDirPaths};
 use aptos_db_indexer::db_indexer::InternalIndexerDB;
@@ -21,6 +35,11 @@ mod aptosdb_test;
 #[cfg(any(test, feature = "fuzzing"))]
 pub mod test_helper;
 
+/// Compile-time gate for the native-position subsystem. Flip to `true`
+/// once order/collateral land and the subsystem ships end-to-end.
+/// Until then, every AptosDB skips opening the position DBs.
+pub(crate) const ENABLE_NATIVE_POSITION: bool = false;
+
 /// This holds a handle to the underlying DB responsible for physical storage and provides APIs for
 /// access to the core Aptos data structures.
 pub struct AptosDB {
@@ -37,6 +56,105 @@ pub struct AptosDB {
     /// This is just to detect concurrent calls to `commit_ledger()`
     commit_lock: std::sync::Mutex<()>,
     update_subscriber: Option<Sender<(Instant, Version)>>,
+    /// Native-position subsystem. `None` until `init_native_position`
+    /// is called from the node-open path. Each future native-mirror
+    /// entity (order, collateral, ...) gets its own bundle field.
+    pub(crate) position: Option<Arc<PositionBundle>>,
+}
+
+/// Durable + runtime handles for the native-position subsystem.
+/// Bundled so the AptosDB struct grows by one field per entity
+/// instead of per handle, and so consumers always reach handles
+/// through a single coherent unit.
+pub struct PositionBundle {
+    pub kv_db: Arc<PositionDb>,
+    pub merkle_db: Arc<PositionMerkleDb>,
+    /// `None` in readonly mode; the async commit pipeline is the
+    /// only piece that isn't safe to construct against a readonly
+    /// mount.
+    pub(crate) state_store: Option<Arc<PositionStateStore>>,
+}
+
+impl AptosDB {
+    /// `Some` once `init_native_position` has wired the subsystem.
+    pub fn position(&self) -> Option<&Arc<PositionBundle>> {
+        self.position.as_ref()
+    }
+
+    /// Build a ready-to-use [`NativeStateCommitter`] from the
+    /// attached bundle. Returns `None` if the subsystem hasn't been
+    /// initialized.
+    pub fn native_state_committer(&self) -> Option<NativeStateCommitter> {
+        let bundle = self.position.as_ref()?;
+        Some(NativeStateCommitter::new(bundle.kv_db.clone()))
+    }
+
+    /// Initialize the native-position subsystem. Opens `position_db` /
+    /// `position_merkle_db` and (in non-readonly mode) spawns the async
+    /// commit pipeline. Idempotent — a second call returns
+    /// `AlreadyExists`. Gated by the `ENABLE_NATIVE_POSITION` compile-
+    /// time constant; called automatically from `open_internal`.
+    pub fn init_native_position(
+        &mut self,
+        db_paths: &StorageDirPaths,
+        rocksdb_config: aptos_config::config::RocksdbConfig,
+        readonly: bool,
+    ) -> Result<()> {
+        if self.position.is_some() {
+            return Err(AptosDbError::Other(
+                "init_native_position called twice; native-position subsystem is already \
+                 attached to this AptosDB"
+                    .to_string(),
+            ));
+        }
+
+        let env = aptos_schemadb::Env::new()
+            .map_err(|e| AptosDbError::Other(format!("failed to create RocksDB env: {e}")))?;
+
+        let position_db = PositionDb::new(db_paths, rocksdb_config, Some(&env), None, readonly)?;
+        if !readonly && let Some(progress) = get_position_commit_progress(&position_db)? {
+            truncate_position_db_shards(&position_db, progress)?;
+        }
+
+        let merkle_db = PositionMerkleDb::new(
+            db_paths,
+            rocksdb_config,
+            Some(&env),
+            None,
+            readonly,
+            /* max_nodes_per_lru_cache_shard — caches off for now */ 0,
+        )?;
+        if !readonly && let Some(progress) = get_position_merkle_commit_progress(&merkle_db)? {
+            truncate_position_merkle_db_shards(&merkle_db, progress)?;
+        }
+        let kv_db = Arc::new(position_db);
+        let merkle_db = Arc::new(merkle_db);
+
+        let state_store = if readonly {
+            None
+        } else {
+            let last_snapshot = new_empty_position_state();
+            Some(Arc::new(PositionStateStore::new_at_snapshot(
+                Arc::clone(&merkle_db),
+                Arc::clone(&self.ledger_db),
+                last_snapshot,
+            )))
+        };
+
+        self.position = Some(Arc::new(PositionBundle {
+            kv_db,
+            merkle_db,
+            state_store,
+        }));
+
+        info!(
+            num_shards = NUM_NATIVE_VALUE_SHARDS,
+            readonly = readonly,
+            "Native-position subsystem initialized."
+        );
+
+        Ok(())
+    }
 }
 
 // DbReader implementations and private functions used by them.
@@ -212,6 +330,13 @@ impl AptosDB {
             cp_path.as_ref(),
             /* is_hot = */ false,
         )?;
+        // Native-position DBs. The static `create_checkpoint` opens
+        // the source DB from `db_path` (creating it if absent), then
+        // checkpoints into `cp_path`. Deployments that never activated
+        // native-position still produce empty position checkpoints —
+        // matches state's always-create behavior.
+        PositionDb::create_checkpoint(db_path.as_ref(), cp_path.as_ref())?;
+        PositionMerkleDb::create_checkpoint(db_path.as_ref(), cp_path.as_ref())?;
 
         info!(
             db_path = db_path.as_ref(),