Custom Indexer and Walrus

A custom Sui indexer paired with Walrus lets you build rich, queryable applications on top of decentralized content-addressable storage. By tracking Metadata dynamic fields onchain, your indexer maintains a local database of blob attributes, such as publisher addresses, view counts, titles, and more, without requiring users to query the Sui full node directly for every read.

Walrus is a decentralized storage and data availability protocol designed specifically for large binary files. It is a content-addressable storage protocol, meaning data is identified and retrieved using a unique identifier called a blob. The protocol derives blobs from the content itself rather than from a file path or location. Consequently, if different users upload the same content, Walrus reuses the existing blob rather than creating a new one.

For uniqueness, each blob uploaded to Walrus also creates a corresponding Blob object on Sui with a unique ID. Furthermore, the associated Blob object can optionally have a Metadata dynamic field.

Metadata dynamic fields extend an onchain object's data at runtime. If set, this dynamic field acts as a mapping of key-value attribute pairs.

You can use the custom indexer framework to extend the existing functionality of Walrus.

info

The Walrus Foundation operates and controls Walrus. For the most accurate and up-to-date information on the Walrus protocol, consult the official Walrus Docs.

Blog platform using Walrus

Sui derives the ID of a dynamic field from its type and parent object's ID. Each Metadata dynamic field ID is also unique. You can leverage these unique characteristics and the Metadata attribute pairs to build a blog platform that enables users to:

• Upload blog posts with titles.
• View their own posts and metrics.
• Delete posts they created.
• Edit post titles.
• Browse posts by other publishers.

Assume a blog platform service already exists to handle uploads to Walrus. When the service creates a blob and its associated Blob object on Sui, it also attaches a Metadata dynamic field containing key-value pairs for publisher (the Sui Address that uploaded the blob), view_count, and title. The service prevents users from modifying the publisher and view_count pairs, but allows the publisher to update the title value.

When a user views a post, the service retrieves the relevant blog post Metadata from the indexed database. It then uses the indexed blob_obj_id value to fetch the Blob object from the full node. The service uses the liveness of the Blob object on Sui to represent whether a blog post is available. If the Blob object is wrapped or deleted, the blog post is not accessible through the service, even if the underlying content on Walrus still exists.

Data modeling

One option for data modeling is to use a single table that maps publisher addresses to Metadata dynamic fields. With this approach, the table is keyed on dynamic_field_id because it both identifies your app data and uniquely represents the content of each uploaded blob.

For example, the up.sql file to create this table might look like the following:

examples/rust/walrus-attributes-indexer/migrations/YYYY-MM-DD-HHMMSS_blog-post/up.sql

-- This table maps a blob to its associated Sui Blob object and the latest dynamic field metadata
-- for traceability. The `view_count` is indexed to serve reads on the app.
CREATE TABLE IF NOT EXISTS blog_post (
    -- ID of the Metadata dynamic field.
    dynamic_field_id            BYTEA         NOT NULL,
    -- Current version of the Metadata dynamic field.
    df_version                  BIGINT        NOT NULL,
    -- Address that published the Walrus Blob.
    publisher                   BYTEA         NOT NULL,
    -- ID of the Blob object on Sui, used during reads to fetch the actual blob content. If this
    -- object has been wrapped or deleted, it will not be present on the live object set, which
    -- means the corresponding content on Walrus is also not accessible.
    blob_obj_id                 BYTEA         NOT NULL,
    view_count                  BIGINT        NOT NULL,
    title                       TEXT          NOT NULL,
    PRIMARY KEY (dynamic_field_id)
);

-- Index to support ordering and filtering by title
CREATE INDEX IF NOT EXISTS blog_post_by_title ON blog_post
(publisher, title);

Reads

To load blog posts from a particular publisher, pass the publisher and LIMIT values to the following query pattern:

SELECT *
FROM blog_post
WHERE publisher = $1
ORDER BY title
LIMIT $2;

Custom indexer implementation

This example uses a sequential pipeline, ensuring each checkpoint is committed once in strict order and as a single atomic operation. The sequential pipeline architecture is not required for this project, but it is a more straightforward option than implementing the concurrent architecture. You can always scale up to the concurrent pipeline if and when your project requires it.

This implementation tracks the latest object state at checkpoint boundary. When the Metadata dynamic field is created, mutated, wrapped or deleted, or unwrapped, it appears among the transaction output under the object changes. You can see an example transaction on Testnet that creates the field. Dynamic fields use the type 0x2::dynamic_field::Field<vector, 0xabc...123::metadata::Metadata>.

Object change to Metadata dynamic field	Included in input objects	Included in live output objects	How to index
Creation (or unwrap)	❌	✅	Insert row
Mutation	✅	✅	Update row
Deletion (or wrap)	✅	❌	Delete row

Processor

All pipelines implement the same Processor trait, which defines the logic to transform a checkpoint from the ingestion task into an intermediate or final form to commit to the store. Data flows into and out of the processor, potentially out of order.

process function

The process function computes the checkpoint_input_objects and latest_live_output_objects sets to capture the state of objects entering and exiting a checkpoint. A Metadata dynamic field that appears in checkpoint_input_objects but not in latest_live_output_objects means it has been either wrapped or deleted. In those cases, you need to record only the dynamic field ID for the commit function to handle later deletion. For creation, mutation, and unwrap operations, the objects always appear in at least the latest_live_output_objects set.

examples/rust/walrus-attributes-indexer/src/handlers/blog_post.rs

/// This pipeline operates on a checkpoint granularity to produce a set of values reflecting the
/// state of relevant Metadata dynamic fields at checkpoint boundary.
async fn process(&self, checkpoint: &Arc<Checkpoint>) -> Result<Vec<Self::Value>> {
    let checkpoint_input_objects = checkpoint_input_objects(checkpoint)?;
    let latest_live_output_objects = checkpoint_output_objects(checkpoint)?;
    // Collect values to be passed to committer. This map is keyed on the dynamic field id.
    let mut values: BTreeMap<ObjectID, Self::Value> = BTreeMap::new();

    // Process relevant Metadata dynamic fields that were wrapped or deleted in this checkpoint.
    for (object_id, object) in &checkpoint_input_objects {
        // If an object appears in both maps, it is still alive at the end of the checkpoint.
        if latest_live_output_objects.contains_key(object_id) {
            continue;
        }

        // Check the checkpoint input state of the Metadata dynamic field to see if it's
        // relevant to our indexing.
        let Some((_, _)) = extract_content_from_metadata(&self.metadata_type, object)? else {
            continue;
        };

        // Since the table is keyed on the dynamic field id, this is all the information we need
        // to delete the correct entry in the commit fn.
        values.insert(*object_id, ProcessedWalrusMetadata::Delete(*object_id));
    }

    for (object_id, object) in &latest_live_output_objects {
        let Some((blog_post_metadata, blob_obj_id)) =
            extract_content_from_metadata(&self.metadata_type, object)?
        else {
            continue;
        };

        values.insert(
            *object_id,
            ProcessedWalrusMetadata::Upsert {
                df_version: object.version().into(),
                dynamic_field_id: *object_id,
                blog_post_metadata,
                blob_obj_id,
            },
        );
    }

    Ok(values.into_values().collect())
}

Committer

The second and final part of the sequential pipeline is the Committer. Because data flows from the processor into the committer out of order, it is the committer's responsibility to batch and write the transformed data to the store in order on checkpoint boundaries.

batch

The batch function defines how to batch transformed data from other processed checkpoints. This function maintains a mapping of dynamic_field_id to the processed Walrus Metadata. The batch function guarantees that the next checkpoint to batch is the next contiguous checkpoint, which means it's safe for you to overwrite the existing entry.

examples/rust/walrus-attributes-indexer/src/handlers/blog_post.rs

fn batch(&self, batch: &mut Self::Batch, values: std::vec::IntoIter<Self::Value>) {
    for value in values {
        match value {
            ProcessedWalrusMetadata::Upsert {
                dynamic_field_id, ..
            } => {
                batch.insert(dynamic_field_id, value);
            }
            ProcessedWalrusMetadata::Delete(dynamic_field_id) => {
                batch.insert(dynamic_field_id, value);
            }
        }
    }
}

commit

The commit function conducts final transformations to the processed data before writing to the store. In this case, the logic partitions the processed data into to_delete and to_upsert.

examples/rust/walrus-attributes-indexer/src/handlers/blog_post.rs

async fn commit<'a>(
    &self,
    batch: &Self::Batch,
    conn: &mut postgres::Connection<'a>,
) -> Result<usize> {
    // Partition the batch into items to delete and items to upsert
    let (to_delete, to_upsert): (Vec<_>, Vec<_>) = batch
        .values()
        .partition(|item| matches!(item, ProcessedWalrusMetadata::Delete(_)));

    let to_upsert: Vec<StoredBlogPost> = to_upsert
        .into_iter()
        .map(|item| item.to_stored())
        .collect::<Result<Vec<_>>>()?;

    let to_delete: Vec<ObjectID> = to_delete
        .into_iter()
        .map(|item| Ok(item.dynamic_field_id()))
        .collect::<Result<Vec<_>>>()?;

    let mut total_affected = 0;

    if !to_delete.is_empty() {
        let deleted_count = diesel::delete(blog_post::table)
            .filter(blog_post::dynamic_field_id.eq_any(to_delete.iter().map(|id| id.to_vec())))
            .execute(conn)
            .await?;

        total_affected += deleted_count;
    }

    if !to_upsert.is_empty() {
        let upserted_count = diesel::insert_into(blog_post::table)
            .values(&to_upsert)
            .on_conflict(blog_post::dynamic_field_id)
            .do_update()
            .set((
                blog_post::df_version.eq(excluded(blog_post::df_version)),
                blog_post::title.eq(excluded(blog_post::title)),
                blog_post::view_count.eq(excluded(blog_post::view_count)),
                blog_post::blob_obj_id.eq(excluded(blog_post::blob_obj_id)),
            ))
            .filter(blog_post::df_version.lt(excluded(blog_post::df_version)))
            .execute(conn)
            .await?;

        total_affected += upserted_count;
    }

    Ok(total_affected)
}

Putting it all together

The main function for the service

examples/rust/walrus-attributes-indexer/src/main.rs

#[tokio::main]
async fn main() -> Result<()> {
    let args = WalrusIndexerArgs::parse();

    // The `IndexerClusterBuilder` offers a convenient way to quickly set up an `IndexerCluster`,
    // which consists of the base indexer, metrics service, and a cancellation token.
    let mut indexer = IndexerClusterBuilder::new()
        .with_database_url(args.database_url)
        .with_args(args.cluster_args)
        .with_migrations(&MIGRATIONS)
        .build()
        .await?;

    let blog_post_pipeline = BlogPostPipeline::new(METADATA_DYNAMIC_FIELD_TYPE).unwrap();

    // Other pipelines can be easily added with `.sequential_pipeline()` or
    // `.concurrent_pipeline()`.
    indexer
        .sequential_pipeline(blog_post_pipeline, SequentialConfig::default())
        .await?;

    match indexer.run().await?.main().await {
        Ok(()) | Err(Error::Terminated) => Ok(()),
        Err(Error::Aborted) => {
            bail!("Indexer aborted due to an unexpected error")
        }
        Err(Error::Task(e)) => {
            bail!(e)
        }
    }
}

To provide users with a list of posts written by a publisher, your service first queries the database on publisher, yielding a result like the following. The service then uses the blob_obj_id to fetch the Blob object contents. From there, you can retrieve the actual Walrus content.

                          dynamic_field_id                          | df_version |                             publisher                              |                            blob_obj_id                             | view_count |      title
--------------------------------------------------------------------+------------+--------------------------------------------------------------------+--------------------------------------------------------------------+------------+------------------
 \x40b5ae12e780ae815d7b0956281291253c02f227657fe2b7a8ccf003a5f597f7 |  608253371 | \xfe9c7a465f63388e5b95c8fd2db857fad4356fc873f96900f4d8b6e7fc1e760e | \xcfb3d474c9a510fde93262d4b7de66cad62a2005a54f31a63e96f3033f465ed3 |         10 | Blog Post Module

How does the indexer connect Walrus blobs to onchain metadata?

Each Walrus blob has a corresponding Blob object on Sui. When your service uploads content to Walrus, it attaches a Metadata dynamic field to that Blob object. The dynamic field stores key-value pairs, such as publisher, view_count, and title, that your indexer tracks across checkpoints.

The indexer monitors object changes in each checkpoint. When a Metadata dynamic field appears in the checkpoint output objects but not in the input objects, the field was just created. When it appears in both, it was mutated. When it appears only in the input objects, it was deleted or wrapped. The sequential pipeline commits each checkpoint atomically, so your database always reflects a consistent state.

Your indexed rows store blob_obj_id, the ID of the Sui Blob object. A Walrus aggregator reads blob content directly from that object ID, so you do not need to resolve the blob_id from the Blob object contents first:

$ curl "$AGGREGATOR/v1/blobs/by-object-id/$BLOB_OBJ_ID" -o post.md

Set $BLOB_OBJ_ID to the blob_obj_id value of a row returned by the publisher query in Putting it all together.

Set $AGGREGATOR to an aggregator endpoint. The Mysten Labs reference endpoint on Mainnet is https://aggregator.walrus-mainnet.walrus.space, and on Testnet it is https://aggregator.walrus-testnet.walrus.space. Do not hardcode a single community endpoint in production. For the complete endpoint list and the rest of the HTTP API, see the Walrus network reference and Reading Blobs with the HTTP API.

Why use a custom indexer instead of querying the full node directly?

Querying a Sui full node for every read operation does not scale well for a blog platform with many publishers and posts. A custom indexer solves this by maintaining a local, queryable database of the onchain state you care about.

The indexer processes each checkpoint once and writes the results to a PostgreSQL database (or another store). Your service then runs fast SQL queries against that database rather than making repeated RPC calls. This separation also lets you add application-specific query patterns, such as sorting posts by title, filtering by publisher, and paginating results, without constructing complex onchain queries.

The sequential pipeline architecture is well suited to this use case because blog post state changes are low-frequency and order matters for consistency. If your platform grows to require higher throughput, you can migrate to the concurrent pipeline without changing your data model.

Why does blob lifetime tracking require additional pipelines?

All Walrus blobs carry an associated lifetime, so you must track expiration changes. A Metadata dynamic field is a child of the Blob object, and you can read the blob's lifetime directly from the Blob object contents. Lifetime changes, however, usually occur on the Blob object itself, and updates to a parent object don't affect the child dynamic field unless you directly modify the child. Because the indexer only watches changes to the Metadata dynamic field, these lifetime changes remain hidden in the current indexing setup.

You can address this in several ways:

• Watch all Blob object changes.
• Watch all BlobCertified events.
• Construct PTBs that make calls to manage blob lifetime and ping the Metadata dynamic field in the same transaction, so both appear in the same checkpoint output.

The third option requires coordinating your write path but keeps the indexer simpler. If you don't want to perform additional work on the write side, then you are limited to the first two options. This requires 2 pipelines running in parallel, one to index metadata as described in the previous sections, and another to index BlobCertified events (or Blob object changes).