refactor(worker): remove AsmState umbrella, persist AnchorState directly

bin/asm-runner/src/worker_context.rs

@@ -9,10 +9,9 @@ use std::sync::Arc;
 use asm_storage::{SledAsmAuxDataDb, SledAsmManifestDb, SledAsmManifestMmrDb, SledAsmStateDb};
 use bitcoin::{Block, BlockHash, Network, block::Header};
 use bitcoind_async_client::{Client, error::ClientError, traits::Reader};
-use strata_asm_common::{AsmLogEntry, AsmManifest, AsmManifestHash, AuxData};
+use strata_asm_common::{AnchorState, AsmManifest, AsmManifestHash, AuxData};
 use strata_asm_worker::{
-    AnchorStateStore, AsmState, AuxDataStore, L1DataProvider, ManifestMmrStore, WorkerError,
-    WorkerResult,
+    AnchorStateStore, AuxDataStore, L1DataProvider, ManifestMmrStore, WorkerError, WorkerResult,
 };
 use strata_btc_types::{BitcoinTxid, L1BlockIdBitcoinExt, RawBitcoinTx};
 use strata_identifiers::{L1BlockCommitment, L1BlockId};
@@ -60,21 +59,6 @@ impl AsmWorkerContext {
             mmr_db,
         }
     }
-
-    /// Loads the STF logs for `blockid` from the manifest store.
-    ///
-    /// The anchor state DB persists only the `AnchorState`, so the logs the STF
-    /// emitted are recovered from the block's manifest. Returns an empty vec
-    /// when no manifest is stored — e.g. the genesis anchor, seeded without
-    /// running the STF.
-    fn manifest_logs(&self, blockid: &L1BlockCommitment) -> WorkerResult<Vec<AsmLogEntry>> {
-        Ok(self
-            .manifest_db
-            .get(blockid)
-            .map_err(|_| WorkerError::DbError)?
-            .map(|manifest| manifest.logs().to_vec())
-            .unwrap_or_default())
-    }
 }
 
 impl L1DataProvider for AsmWorkerContext {
@@ -180,15 +164,11 @@ impl L1DataProvider for AsmWorkerContext {
 }
 
 impl AnchorStateStore for AsmWorkerContext {
-    // The state store persists only the `AnchorState`; the worker's `AsmState`
-    // umbrella also carries the STF logs, which live in the manifest store.
-    // Reads rejoin the two so the reconstructed `AsmState` matches what the STF
-    // produced — anything that derives from the logs (the MohoState, the
-    // export-entry index) then stays correct even when it runs over a reloaded
-    // state rather than fresh STF output. Returning empty logs here once let a
-    // re-committed anchor silently drop a block's export entries and predicate
-    // update, desyncing its persisted MohoState from the proven one.
-    fn get_latest_asm_state(&self) -> WorkerResult<Option<(L1BlockCommitment, AsmState)>> {
+    // The state store persists the `AnchorState` on its own; the STF logs live
+    // in the manifest store and every consumer that needs them (the Moho
+    // worker's `get_anchor_logs`, the checkpoint/bridge test harness) reads them
+    // from there directly, keyed by block.
+    fn get_latest_asm_state(&self) -> WorkerResult<Option<(L1BlockCommitment, AnchorState)>> {
         let Some(anchor) = self
             .state_db
             .get_latest()
@@ -197,28 +177,22 @@ impl AnchorStateStore for AsmWorkerContext {
             return Ok(None);
         };
         let blockid = anchor.chain_view.pow_state.last_verified_block;
-        let logs = self.manifest_logs(&blockid)?;
-        Ok(Some((blockid, AsmState::new(anchor, logs))))
+        Ok(Some((blockid, anchor)))
     }
 
-    fn get_anchor_state(&self, blockid: &L1BlockCommitment) -> WorkerResult<AsmState> {
-        let anchor = self
-            .state_db
+    fn get_anchor_state(&self, blockid: &L1BlockCommitment) -> WorkerResult<AnchorState> {
+        self.state_db
             .get(blockid)
             .map_err(|_| WorkerError::DbError)?
-            .ok_or(WorkerError::MissingAsmState(*blockid.blkid()))?;
-        let logs = self.manifest_logs(blockid)?;
-        Ok(AsmState::new(anchor, logs))
+            .ok_or(WorkerError::MissingAsmState(*blockid.blkid()))
     }
 
     fn store_anchor_state(
         &self,
         _blockid: &L1BlockCommitment,
-        state: &AsmState,
+        state: &AnchorState,
     ) -> WorkerResult<()> {
-        self.state_db
-            .put(state.state())
-            .map_err(|_| WorkerError::DbError)?;
+        self.state_db.put(state).map_err(|_| WorkerError::DbError)?;
 
         Ok(())
     }

crates/extensions/moho/worker/src/state.rs

@@ -19,7 +19,7 @@ use crate::{MohoWorkerContext, MohoWorkerResult, compute, constants};
 /// state in the store. It keeps no chain view of its own.
 ///
 /// Mirrors `strata-asm-worker`'s `AsmWorkerServiceState`, which likewise holds
-/// the current `AsmState` in memory and re-anchors on reorg. The fold
+/// the current `AnchorState` in memory and re-anchors on reorg. The fold
 /// orchestration lives in the service layer's `process_block`; this type just
 /// holds the data and the small `update_moho_state` mutation that advances it.
 #[derive(Debug)]

crates/storage/src/state.rs

@@ -1,9 +1,8 @@
 //! Storage trait for ASM anchor states.
 //!
 //! Each entry records the [`AnchorState`] computed after processing the L1 block
-//! identified by the given [`L1BlockCommitment`]. The worker's `AsmState`
-//! umbrella (anchor state plus logs) is deliberately not stored here: only the
-//! anchor state is persistent state; the logs live in the manifest store.
+//! identified by the given [`L1BlockCommitment`]. Only the anchor state is
+//! persistent state; the STF logs live in the manifest store.
 
 use std::fmt::Debug;
 

crates/worker/Cargo.toml

@@ -22,7 +22,6 @@ strata-merkle-node-store = { workspace = true, optional = true }
 
 anyhow.workspace = true
 bitcoin.workspace = true
-borsh.workspace = true
 futures.workspace = true
 serde.workspace = true
 thiserror.workspace = true

crates/worker/src/lib.rs

@@ -3,7 +3,6 @@
 //! The `strata-asm-worker` crate provides a dedicated asynchronous worker
 //! for managing Strata's Anchor state (ASM).
 
-mod asm_state;
 mod aux_resolver;
 mod builder;
 mod constants;
@@ -18,7 +17,6 @@ mod sync;
 pub mod test_utils;
 mod traits;
 
-pub use asm_state::AsmState;
 pub use aux_resolver::AuxDataResolver;
 pub use builder::AsmWorkerBuilder;
 pub use errors::{AnchorMismatch, WorkerError, WorkerResult};

crates/worker/src/message.rs

@@ -1,15 +1,16 @@
 //! Messages from the handle to the worker.
 
 use bitcoin::BlockHash;
+use strata_asm_common::AnchorState;
 use strata_identifiers::L1BlockCommitment;
 use strata_service::CommandCompletionSender;
 
-use crate::{AsmState, WorkerResult};
+use crate::WorkerResult;
 
 /// Messages from the ASM Handle to the subprotocol to give it work to do.
 #[derive(Debug)]
 pub enum SubprotocolMessage {
-    NewAsmState(AsmState, L1BlockCommitment),
+    NewAsmState(AnchorState, L1BlockCommitment),
 }
 
 /// Messages from the handle to the ASM worker, with a completion sender to

crates/worker/src/service.rs

@@ -3,15 +3,15 @@
 use std::marker;
 
 use serde::{Deserialize, Serialize};
-use strata_asm_common::AsmSpec;
+use strata_asm_common::{AnchorState, AsmSpec};
 use strata_btc_types::BlockHashExt;
 use strata_identifiers::{L1BlockCommitment, L1BlockId};
 use strata_service::{Response, Service, SyncService};
 use tracing::*;
 
 use crate::{
-    AsmState, AsmWorkerServiceState, SyncError, SyncPlan, WorkerError, message::AsmWorkerMessage,
-    plan_sync, traits::WorkerContext,
+    AsmWorkerServiceState, SyncError, SyncPlan, WorkerError, message::AsmWorkerMessage, plan_sync,
+    traits::WorkerContext,
 };
 
 /// ASM service implementation using the service framework.
@@ -86,7 +86,7 @@ where
 /// lower height). Runs in two phases:
 ///
 /// 1. **Plan** (backward): from `target`, follow parent links via each block's `prev_blockhash`
-///    back to the base — the most recent ancestor with a stored `AsmState` — collecting the
+///    back to the base — the most recent ancestor with a stored `AnchorState` — collecting the
 ///    unprocessed blocks in between. This walks `target`'s own ancestry, so on an L1 reorg the base
 ///    is the fork point and the abandoned branch is never visited. Only block headers are read
 ///    here, so a deep reorg does not load every intervening block into memory at once. See
@@ -198,7 +198,7 @@ where
 }
 
 /// Walks back from `target` along parent links to build a
-/// [`SyncPlan<AsmState>`]: the base — the most recent ancestor with a stored
+/// [`SyncPlan<AnchorState>`]: the base — the most recent ancestor with a stored
 /// anchor state — and the unprocessed blocks between it and `target`.
 ///
 /// A thin adapter over [`plan_sync`]: the base is a stored anchor state, and
@@ -211,7 +211,7 @@ fn plan_block_processing<W: WorkerContext>(
     ctx: &W,
     target: &L1BlockCommitment,
     genesis_height: u64,
-) -> crate::WorkerResult<SyncPlan<AsmState>> {
+) -> crate::WorkerResult<SyncPlan<AnchorState>> {
     plan_sync(
         *target,
         genesis_height,
@@ -274,8 +274,9 @@ where
     state.context.store_aux_data(block_id, &aux_data)?;
 
     // Anchor state last: it is the block's commit point (see fn docs), so a
-    // crash before it leaves the block uncommitted to be safely re-run.
-    let new_state = AsmState::from_output(asm_stf_out);
+    // crash before it leaves the block uncommitted to be safely re-run. The
+    // STF's logs are already persisted in the manifest recorded above.
+    let new_state = asm_stf_out.state;
     state.context.store_anchor_state(block_id, &new_state)?;
     state.update_anchor_state(new_state, *block_id);
 
@@ -292,19 +293,7 @@ where
 pub struct AsmWorkerStatus {
     pub is_initialized: bool,
     pub cur_block: Option<L1BlockCommitment>,
-    pub cur_state: Option<AsmState>,
-}
-
-impl AsmWorkerStatus {
-    /// Get the logs from the current ASM state.
-    ///
-    /// Returns an empty slice if the state is not initialized.
-    pub fn logs(&self) -> &[strata_asm_common::AsmLogEntry] {
-        self.cur_state
-            .as_ref()
-            .map(|s| s.logs().as_slice())
-            .unwrap_or(&[])
-    }
+    pub cur_state: Option<AnchorState>,
 }
 
 #[cfg(test)]
@@ -331,19 +320,14 @@ mod tests {
     /// the snapshot size [`AsmWorkerServiceState::transition`] resolves aux data
     /// against.
     fn anchor_leaf_count(state: &AsmWorkerServiceState<TestAsmWorkerContext, TestAsmSpec>) -> u64 {
-        state
-            .anchor
-            .state()
-            .chain_view
-            .history_accumulator
-            .num_entries()
+        state.anchor.chain_view.history_accumulator.num_entries()
     }
 
     /// Pending block heights in the order they're processed (oldest first).
     ///
     /// `plan.pending` is stored newest-first; reversing here keeps the test
     /// expectations ascending, which is easier to read.
-    fn pending_heights(plan: &SyncPlan<AsmState>) -> Vec<u32> {
+    fn pending_heights(plan: &SyncPlan<AnchorState>) -> Vec<u32> {
         plan.pending.iter().rev().map(|b| b.height()).collect()
     }
 

crates/worker/src/state.rs

@@ -1,5 +1,5 @@
 use bitcoin::{Block, CompactTarget, params::Params};
-use strata_asm_common::{AsmSpec, AuxData, HeaderVerificationState};
+use strata_asm_common::{AnchorState, AsmSpec, AuxData, HeaderVerificationState};
 use strata_asm_stf::AsmStfOutput;
 use strata_btc_types::BlockHashExt;
 use strata_btc_verification::{
@@ -10,8 +10,8 @@ use strata_service::ServiceState;
 use tracing::field::Empty;
 
 use crate::{
-    AnchorMismatch, AsmState, L1DataProvider, Subscribers, WorkerContext, WorkerError,
-    WorkerResult, aux_resolver::AuxDataResolver, constants,
+    AnchorMismatch, L1DataProvider, Subscribers, WorkerContext, WorkerError, WorkerResult,
+    aux_resolver::AuxDataResolver, constants,
 };
 
 /// Service state for the ASM worker.
@@ -27,8 +27,8 @@ pub struct AsmWorkerServiceState<W, S: AsmSpec> {
     /// ASM spec driving the subprotocol pipeline.
     pub(crate) spec: S,
 
-    /// Current ASM state.
-    pub anchor: AsmState,
+    /// Current ASM anchor state.
+    pub anchor: AnchorState,
 
     /// Current anchor block.
     pub blkid: L1BlockCommitment,
@@ -86,9 +86,8 @@ where
                 let genesis_blk = genesis_state.chain_view.pow_state.last_verified_block;
                 tracing::info!(%genesis_blk, "no stored ASM state; initializing genesis anchor");
 
-                let state = AsmState::new(genesis_state, vec![]);
-                context.store_anchor_state(&genesis_blk, &state)?;
-                (state, genesis_blk)
+                context.store_anchor_state(&genesis_blk, &genesis_state)?;
+                (genesis_state, genesis_blk)
             }
         };
 
@@ -118,7 +117,7 @@ where
             let span = tracing::debug_span!("asm.stf.pre_process", protocol_txs = Empty);
             let _guard = span.enter();
 
-            let result = strata_asm_stf::pre_process_asm(&self.spec, cur_state.state(), block)
+            let result = strata_asm_stf::pre_process_asm(&self.spec, cur_state, block)
                 .map_err(WorkerError::AsmError)?;
 
             span.record("protocol_txs", result.txs.len());
@@ -133,7 +132,7 @@ where
             // Snapshot proofs at the accumulator's own leaf count: a verifier
             // checks them against this accumulator's committed root, so the
             // snapshot size must be that accumulator's.
-            let accumulator = &cur_state.state().chain_view.history_accumulator;
+            let accumulator = &cur_state.chain_view.history_accumulator;
             let resolver = AuxDataResolver::new(&self.context, accumulator.num_entries());
             resolver.resolve(&pre_process.aux_requests)?
         };
@@ -146,7 +145,7 @@ where
 
         strata_asm_stf::compute_asm_transition(
             &self.spec,
-            cur_state.state(),
+            cur_state,
             block,
             &aux_data,
             Some(&coinbase_inclusion_proof),
@@ -156,7 +155,7 @@ where
     }
 
     /// Updates anchor related bookkeeping.
-    pub(crate) fn update_anchor_state(&mut self, anchor: AsmState, blkid: L1BlockCommitment) {
+    pub(crate) fn update_anchor_state(&mut self, anchor: AnchorState, blkid: L1BlockCommitment) {
         self.anchor = anchor;
         self.blkid = blkid;
     }