Unbounded Growth in COL_BLOB_STATUS Due to Missing Pruning Logic

[EFCCWEB3]

Brief/Intro

Your 0G DA node implements a pruning mechanism that removes old blob data from COL_SLICE based on epoch windows, but fails to prune corresponding status entries from COL_BLOB_STATUS, causing unbounded database growth and state divergence. This creates a scenario where the database tracks blob statuses for epochs whose actual data has been deleted, leading to wasted storage and potential confusion during blob signing operations. While not a direct security vulnerability, this design flaw impacts node operational efficiency and database consistency over time.

Vulnerability Details

Architecture Overview

The node maintains two separate progress trackers in COL_MISC:

• SYNC_PROGRESS_KEY (value 0): Tracks block numbers for DA log monitoring
• PRUNE_PROGRESS_KEY (value 1): Tracks epoch numbers for data pruning

const SYNC_PROGRESS_KEY: &[u8] = &[0];
const PRUNE_PROGRESS_KEY: &[u8] = &[1];

here

Execution Flow Analysis

1. Startup Sequence

graph TD
    A[start_server] --> B[setup_chain_state]
    B --> C[start_da_monitor]
    B --> D[start_grpc_server]
    A --> E[start_pruner]
    
    C --> F[spawn DA monitor loop]
    E --> G[spawn pruner task]

Loading

The server starts both the DA monitor and pruner as concurrent Tokio tasks.

async fn start_server(ctx: &Context) -> Result<()> {
    let chain_state = setup_chain_state(ctx).await?;
    start_grpc_server(chain_state.clone(), ctx).await?;
    start_pruner(chain_state.clone(), ctx).await?;
    Ok(())
}

2. Pruner Execution Flow

graph TD
    A[run_pruner] --> B[get_prune_progress]
    B --> C{progress exists?}
    C -->|No| D[put_prune_progress 0]
    C -->|Yes| E[prune loop]
    E --> F[get current_epoch]
    E --> G[get epoch_window_size]
    E --> H[while pruned + 1 + window < current_epoch]
    H --> I[db.prune epoch]
    H --> J[put_prune_progress]
    I --> K[delete_prefix COL_SLICE]

Loading

 async fn prune(&self, epoch: u64) -> Result<()> {
        let blob_prefix: Vec<u8> = once(BLOB_PREFIX).chain(epoch.to_be_bytes()).collect();
        let slice_prefix: Vec<u8> = once(SLICE_PREFIX).chain(epoch.to_be_bytes()).collect();
        let data_prefix: Vec<u8> = once(DATA_PREFIX).chain(epoch.to_be_bytes()).collect();

        let mut tx = self.db.transaction();
        tx.delete_prefix(COL_SLICE, &blob_prefix);
        tx.delete_prefix(COL_SLICE, &slice_prefix);
        tx.delete_prefix(COL_SLICE, &data_prefix);

        self.db.write(tx)?;
        Ok(())
    }

The pruner removes data from COL_SLICE using three prefix deletions:

• BLOB_PREFIX (0): Blob metadata
• SLICE_PREFIX (1): Compressed slice data
• DATA_PREFIX (2): Raw slice data

Critical Gap: No equivalent pruning exists for COL_BLOB_STATUS (column 4) .

3. DA Monitor Execution Flow

graph TD
    A[start_da_monitor] --> B[get_sync_progress]
    B --> C{progress exists?}
    C -->|No| D[put_sync_progress start_block]
    C -->|Yes| E[spawn monitoring loop]
    E --> F[check_da_logs every 5s]
    F --> G[get_sync_progress]
    F --> H[get finalized block]
    F --> I[check_data_logs from-to]
    I --> J[check_data_upload]
    I --> K[check_data_verified]
    J --> L[get_blob_status]
    L --> M{status exists?}
    M -->|No| N[put_blob UPLOADED]
    M -->|Yes| O[skip]

Loading

The DA monitor processes DataUpload events and writes to COL_BLOB_STATUS without checking if the epoch was pruned.

async fn check_data_upload(chain_state: Arc<ChainState>, l: u64, r: u64) -> Result<()> {
    let filter: ethers::types::Filter = chain_state
        .da_entrance
        .data_upload_filter()
        .from_block(l)
        .to_block(r)
        .address(chain_state.da_entrance.address().into())
        .filter;
    for log in chain_state.provider.get_logs(&filter).await? {
        match DataUploadFilter::decode_log(&RawLog {
            topics: log.topics,
            data: log.data.to_vec(),
        }) {
            Ok(event) => {
                let epoch = event.epoch.as_u64();
                let quorum_id = event.quorum_id.as_u64();
                let maybe_blob_status = chain_state
                    .db
                    .read()
                    .await
                    .get_blob_status(epoch, quorum_id, event.data_root)
                    .await?;
                match maybe_blob_status {
                    Some(_) => {}
                    None => {
                        chain_state
                            .db
                            .write()
                            .await
                            .put_blob(epoch, quorum_id, event.data_root, BlobStatus::UPLOADED)
                            .await?;
                        info!(
                            "new file found, epoch: {:?}, quorum_id: {:?}, data_root: {:X?}",
                            epoch, quorum_id, event.data_root
                        );
                    }
                }
            }
            Err(e) => {
                error!("log decode error: e={:?}", e);
            }
        }
    }
    Ok(())

here

4. Zombie Data Creation Scenario

Analogy: Imagine a library that removes old books from shelves (pruning) but keeps the catalog cards claiming those books still exist. Patrons can request "ghost books" that the catalog says exist but aren't actually on shelves.

Step-by-step execution:

1. Node goes offline at block 1000, epoch 50
2. Current epoch advances to 100, epoch_window_size = 10
3. Node restarts:
   • Pruner immediately prunes epochs 1-89 from COL_SLICE
   • DA monitor resumes from block 1000, processes blocks 1000-50000
4. For each DataUpload event in pruned epochs:
   • get_blob_status() returns None (new node scenario)
   • put_blob() writes UPLOADED status to COL_BLOB_STATUS
5. Result: COL_BLOB_STATUS contains entries for epochs 1-89, but COL_SLICE has no data for those epochs

5. Impact on Signing Flow

The batch_sign function checks blob status before signing:

match maybe_blob_status {
    Some(BlobStatus::UPLOADED) => Ok(()),  // Allows zombie blobs
    Some(BlobStatus::VERIFIED) => Err(...),
    None => Err(...),
}

No epoch age validation exists, so requests for ancient epochs with zombie status will pass this check and proceed to signing.

 async fn batch_sign_inner(
        &self,
        request: Request<BatchSignRequest>,
    ) -> Result<Response<BatchSignReply>, Status> {
        let remote_addr = request.remote_addr();
        let request_content = request.into_inner();
        metrics::GRPC_REQ_GAUGE.set(request_content.encoded_len() as f64);
        let ts = Instant::now();

        info!(?remote_addr, "Received sign request");
        let mut reply = BatchSignReply { signatures: vec![] };

        for req in request_content.requests.iter() {
            let (storage_root, erasure_commitment) = Self::decode_root(req)?;

            self.check_blob_status(req, storage_root).await?;

            let encoded_slices = Self::decode_encoded_slices(req)?;

            let res = self
                .verify_encoded_slices(
                    req.epoch,
                    req.quorum_id,
                    storage_root,
                    erasure_commitment,
                    &encoded_slices,
                )
                .await;

            if let Err(error) = res {
                return Err(match error {
                    VerificationError::Internal(e) => Status::new(
                        Code::Internal,
                        format!("internal error on verification: {:?}", e),
                    ),
                    VerificationError::SliceMismatch => Status::new(
                        Code::InvalidArgument,
                        "received slices and assigned slices are mismatch",
                    ),
                    VerificationError::IncorrectSlice(e) => Status::new(
                        Code::InvalidArgument,
                        format!("verification failed: {:?}", e),
                    ),
                    VerificationError::DeferredVerifyFail => Status::new(
                        Code::InvalidArgument,
                        "received slice does not pass pairing check, the accelerated verification algorithm cannot detect the specific error location".to_string(),
                    ),
                });
            }

            let hash =
                blob_verified_hash(storage_root, req.epoch, req.quorum_id, erasure_commitment);
            let signature = (hash * self.signer_bls_private_key).into_affine();
            let mut value = Vec::new();
            signature.serialize_uncompressed(&mut value);
            reply.signatures.push(value);
            // write slices to db
            self.db
                .write()
                .await
                .put_slice(req.epoch, req.quorum_id, storage_root, encoded_slices)
                .await
                .map_err(|e| Status::new(Code::Internal, format!("put slice error: {:?}", e)))?;
        }

        info!("responsed in {:?} ms", ts.elapsed().as_millis());
        Ok(Response::new(reply))
    }

here

Impact Details

Storage Impact

• Unbounded Growth: COL_BLOB_STATUS grows indefinitely - one entry per blob per epoch, forever
• State Divergence: Database tracks blobs for epochs whose data was deleted
• Wasted I/O: Queries scan through obsolete status entries

Recommendation

1. Add Blob Status Pruning: Extend SliceDB::prune() to also delete from COL_BLOB_STATUS:

async fn prune(&self, epoch: u64) -> Result<()> {
    // Existing slice pruning...
    let blob_status_prefix: Vec<u8> = epoch.to_be_bytes().to_vec();
    tx.delete_prefix(COL_BLOB_STATUS, &blob_status_prefix);
    // ... rest of implementation
}

2. Add Epoch Validation: In check_blob_status(), validate epoch is within window:

let current_epoch = get_current_epoch().await?;
if req.epoch + epoch_window_size < current_epoch {
    return Err(Status::new(Code::InvalidArgument, "epoch too old"));
}