feat(indexer): Make concurrent modifications to objects and display table safe

[tomxey]

Description of change

Make concurrent modifications to objects and display table safe.
It's a prerequisite for optimistic indexing, which will run concurrent to standard checkpoint indexing and may introduce race conditions.

Links to any relevant issues

fixes #6320

Type of change

• Enhancement (a non-breaking change which adds functionality)

How the change has been tested

cargo ci-clippy
cargo nextest run -p iota-indexer --features pg_integration --test-threads 1

Tested on local branch with optimistic indexing enabled, which results in concurrent updates of objects and display tables, the following scenarios were tested:

• consecutive updates for single object
• consecutive wraps and unwraps of single object (which effectively deletes and undeletes and object)
• consecutive display updates

No inconsistencies were spotted in the data returned via api when running read queries inbetween the modification requests.

Change checklist

Tick the boxes that are relevant to your changes, and delete any items that are not.

• I have followed the contribution guidelines for this project
• I have performed a self-review of my own code
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have added tests that prove my fix is effective or that my feature works
• I have checked that new and existing unit tests pass locally with my changes

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

4 Skipped Deployments

Name	Status	Preview	Comments	Updated (UTC)
apps-backend	⬜️ Ignored (Inspect)	Visit Preview		Apr 23, 2025 2:01pm
apps-ui-kit	⬜️ Ignored (Inspect)	Visit Preview		Apr 23, 2025 2:01pm
rebased-explorer	⬜️ Ignored (Inspect)	Visit Preview		Apr 23, 2025 2:01pm
wallet-dashboard	⬜️ Ignored (Inspect)	Visit Preview		Apr 23, 2025 2:01pm

[sergiupopescu199]

lgtm 🐙, a minor nit around cloning the query.

[sergiupopescu199]

Suggested change

	for obj in mutated_object_mutation_chunk.iter().cloned() {
	for obj in mutated_object_mutation_chunk {

[sergiupopescu199]

Nit: I'm wondering if cloning here is ok and does not introduce more memory consumption. Even tough if the transactional_blocking_with_retry succeeds at the first try we still have two copies of the query, I suppose the vectors are not small and potentially could increase the usage of the RAM. If cloning is not an concert then you can safely ignore my comment.

I was able to avoid cloning by moving the query inside the transactional_blocking_with_retry! and using references for the vectors.

transactional_blocking_with_retry!(
            &self.blocking_cp,
            |conn| {
                diesel::sql_query(
                    r#"
                        WITH new_data AS (
                            SELECT
                                unnest($1::bytea[])  AS object_id,
                                unnest($2::bigint[]) AS object_version
                        ),
                        locked_objects AS (
                            SELECT o.*
                            FROM objects o
                            JOIN new_data nd ON o.object_id = nd.object_id
                            FOR UPDATE
                        ),
                        locked_deletes AS (
                            SELECT del.*
                            FROM optimistic_deleted_objects_versions del
                            JOIN new_data nd ON del.object_id = nd.object_id
                            FOR UPDATE
                        ),
                        deleted AS (
                            DELETE FROM objects o
                            USING new_data nd
                            WHERE o.object_id = nd.object_id
                              AND nd.object_version > o.object_version
                        )
                        INSERT INTO optimistic_deleted_objects_versions (object_id, object_version)
                        SELECT object_id, object_version
                        FROM new_data
                        ON CONFLICT (object_id)
                        DO UPDATE
                        SET object_version = EXCLUDED.object_version
                        WHERE EXCLUDED.object_version > optimistic_deleted_objects_versions.object_version;
                    "#,
                )
                .bind::<sql_types::Array<sql_types::Bytea>, _>(&object_id_vec)
                .bind::<sql_types::Array<sql_types::BigInt>, _>(&object_version_vec)
                    .execute(conn)
                    .map_err(IndexerError::from)
                    .context("Failed to write object deletion to PostgresDB")?;
                Ok::<(), IndexerError>(())
            },

same approach was applied for the persist_object_mutation_chunk method.

[samuel-rufi]

Please see the failing CI