update pooled blob reference counting

halfprice · halfprice · commit 0a141060eb11 · 2026-03-10T19:31:43.000-07:00
diff --git a/crates/walrus-service/src/node.rs b/crates/walrus-service/src/node.rs
@@ -1576,13 +1576,7 @@ impl StorageNode {
                     .await?;
             }
             EventStreamElement::ContractEvent(ContractEvent::StoragePoolEvent(pool_event)) => {
-                self.process_storage_pool_event(
-                    blob_event_processor,
-                    event_handle,
-                    pool_event,
-                    checkpoint_position,
-                )
-                .await?;
+                self.process_storage_pool_event(event_handle, pool_event)?;
             }
             EventStreamElement::ContractEvent(ContractEvent::DenyListEvent(_event)) => {
                 // TODO: Implement DenyListEvent handling (WAL-424)
@@ -1624,19 +1618,29 @@ impl StorageNode {
     }
 
     #[tracing::instrument(skip_all)]
-    async fn process_storage_pool_event(
+    fn process_storage_pool_event(
         &self,
-        blob_event_processor: &BlobEventProcessor,
         event_handle: EventHandle,
         event: StoragePoolEvent,
-        checkpoint_position: CheckpointEventPosition,
     ) -> anyhow::Result<()> {
         let _scope = monitored_scope::monitored_scope("ProcessEvent::StoragePoolEvent");
 
         tracing::debug!(?event, "{} event received", event.name());
-        blob_event_processor
-            .process_storage_pool_event(event_handle, event, checkpoint_position)
-            .await?;
+        match event {
+            StoragePoolEvent::StoragePoolCreated(ref created) => {
+                self.inner
+                    .storage
+                    .set_storage_pool_end_epoch(&created.storage_pool_id, created.end_epoch)
+                    .context("failed to set storage pool end epoch")?;
+            }
+            StoragePoolEvent::StoragePoolExtended(ref extended) => {
+                self.inner
+                    .storage
+                    .set_storage_pool_end_epoch(&extended.storage_pool_id, extended.new_end_epoch)
+                    .context("failed to update storage pool end epoch")?;
+            }
+        }
+        event_handle.mark_as_complete();
         Ok(())
     }
 
diff --git a/crates/walrus-service/src/node/blob_event_processor.rs b/crates/walrus-service/src/node/blob_event_processor.rs
@@ -3,11 +3,10 @@
 
 use std::{num::NonZeroUsize, sync::Arc};
 
-use anyhow::Context as _;
 use sui_macros::fail_point_async;
 use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender};
 use walrus_core::{BlobId, Epoch};
-use walrus_sui::types::{BlobEvent, InvalidBlobId, StoragePoolEvent};
+use walrus_sui::types::{BlobEvent, InvalidBlobId};
 use walrus_utils::metrics::monitored_scope;
 
 use self::pending_events::{PendingEventCounter, PendingEventGuard};
@@ -416,8 +415,11 @@ impl BlobEventProcessor {
             .storage
             .update_blob_info(event_handle.index(), &blob_event)?;
 
-        if let BlobEvent::Registered(event) = &blob_event {
-            self.handle_registered_event(event_handle, event.blob_id)
+        if matches!(
+            &blob_event,
+            BlobEvent::Registered(_) | BlobEvent::PooledBlobRegistered(_)
+        ) {
+            self.handle_registered_event(event_handle, blob_event.blob_id())
                 .await?;
             return Ok(());
         }
@@ -464,78 +466,6 @@ impl BlobEventProcessor {
         Ok(())
     }
 
-    /// Processes a storage pool event.
-    ///
-    /// For blob-related events (Registered, Certified, Deleted), this updates the blob info and
-    /// storage pool tables synchronously, then delegates to the background processor for
-    /// post-processing (same as regular blob events). For storage-level events (Created,
-    /// Extended), only the storage pool end_epoch table is updated.
-    pub(super) async fn process_storage_pool_event(
-        &self,
-        event_handle: EventHandle,
-        event: StoragePoolEvent,
-        checkpoint_position: CheckpointEventPosition,
-    ) -> anyhow::Result<()> {
-        match event {
-            StoragePoolEvent::StoragePoolCreated(ref created) => {
-                self.node
-                    .storage
-                    .set_storage_pool_end_epoch(&created.storage_pool_id, created.end_epoch)
-                    .context("failed to set storage pool end epoch")?;
-                event_handle.mark_as_complete();
-            }
-            StoragePoolEvent::PooledBlobRegistered(ref registered) => {
-                self.node
-                    .storage
-                    .process_pooled_blob_registered(event_handle.index(), registered)
-                    .context("failed to process pooled blob registered")?;
-
-                self.handle_registered_event(event_handle, registered.blob_id)
-                    .await?;
-            }
-            StoragePoolEvent::PooledBlobCertified(ref certified) => {
-                self.node
-                    .storage
-                    .process_pooled_blob_certified(event_handle.index(), certified)
-                    .context("failed to process pooled blob certified")?;
-
-                let blob_event = BlobEvent::PooledBlobCertified(certified.clone());
-                self.dispatch_task(
-                    blob_event.blob_id(),
-                    BackgroundTask::ProcessEvent {
-                        event_handle,
-                        blob_event,
-                        checkpoint_position,
-                    },
-                )?;
-            }
-            StoragePoolEvent::PooledBlobDeleted(ref deleted) => {
-                self.node
-                    .storage
-                    .process_pooled_blob_deleted(event_handle.index(), deleted)
-                    .context("failed to process pooled blob deleted")?;
-
-                let blob_event = BlobEvent::PooledBlobDeleted(deleted.clone());
-                self.dispatch_task(
-                    blob_event.blob_id(),
-                    BackgroundTask::ProcessEvent {
-                        event_handle,
-                        blob_event,
-                        checkpoint_position,
-                    },
-                )?;
-            }
-            StoragePoolEvent::StoragePoolExtended(ref extended) => {
-                self.node
-                    .storage
-                    .set_storage_pool_end_epoch(&extended.storage_pool_id, extended.new_end_epoch)
-                    .context("failed to update storage pool end epoch")?;
-                event_handle.mark_as_complete();
-            }
-        }
-        Ok(())
-    }
-
     pub fn get_pending_event_counter(&self) -> PendingEventCounter {
         self.background_pending_event_count.clone()
     }
diff --git a/crates/walrus-service/src/node/metrics.rs b/crates/walrus-service/src/node/metrics.rs
@@ -377,9 +377,6 @@ impl TelemetryLabel for StoragePoolEvent {
     fn label(&self) -> &'static str {
         match self {
             StoragePoolEvent::StoragePoolCreated(_) => "storage-pool-created",
-            StoragePoolEvent::PooledBlobRegistered(_) => "pooled-blob-registered",
-            StoragePoolEvent::PooledBlobCertified(_) => "pooled-blob-certified",
-            StoragePoolEvent::PooledBlobDeleted(_) => "pooled-blob-deleted",
             StoragePoolEvent::StoragePoolExtended(_) => "storage-pool-extended",
         }
     }
diff --git a/crates/walrus-service/src/node/storage.rs b/crates/walrus-service/src/node/storage.rs
@@ -683,36 +683,6 @@ impl Storage {
             .set_storage_pool_end_epoch(storage_pool_id, end_epoch)
     }
 
-    /// Processes a `PooledBlobRegistered` event.
-    pub(crate) fn process_pooled_blob_registered(
-        &self,
-        event_index: u64,
-        event: &walrus_sui::types::PooledBlobRegistered,
-    ) -> Result<(), TypedStoreError> {
-        self.blob_info
-            .process_pooled_blob_registered(event_index, event)
-    }
-
-    /// Processes a `PooledBlobCertified` event.
-    pub(crate) fn process_pooled_blob_certified(
-        &self,
-        event_index: u64,
-        event: &walrus_sui::types::PooledBlobCertified,
-    ) -> Result<(), TypedStoreError> {
-        self.blob_info
-            .process_pooled_blob_certified(event_index, event)
-    }
-
-    /// Processes a `PooledBlobDeleted` event.
-    pub(crate) fn process_pooled_blob_deleted(
-        &self,
-        event_index: u64,
-        event: &walrus_sui::types::PooledBlobDeleted,
-    ) -> Result<(), TypedStoreError> {
-        self.blob_info
-            .process_pooled_blob_deleted(event_index, event)
-    }
-
     /// Returns the current event cursor and the next event index.
     #[tracing::instrument(skip_all)]
     pub fn get_event_cursor_and_next_index(
diff --git a/crates/walrus-service/src/node/storage/blob_info.rs b/crates/walrus-service/src/node/storage/blob_info.rs
diff --git a/crates/walrus-sui/src/types/events.rs b/crates/walrus-sui/src/types/events.rs

Original file line number	Diff line number	Diff line change
`@@ -377,9 +377,6 @@ impl TelemetryLabel for StoragePoolEvent {`
`377`	`377`	`fn label(&self) -> &'static str {`
`378`	`378`	`match self {`
`379`	`379`	`StoragePoolEvent::StoragePoolCreated(_) => "storage-pool-created",`
`380`		`- StoragePoolEvent::PooledBlobRegistered(_) => "pooled-blob-registered",`
`381`		`- StoragePoolEvent::PooledBlobCertified(_) => "pooled-blob-certified",`
`382`		`- StoragePoolEvent::PooledBlobDeleted(_) => "pooled-blob-deleted",`
`383`	`380`	`StoragePoolEvent::StoragePoolExtended(_) => "storage-pool-extended",`
`384`	`381`	`}`
`385`	`382`	`}`